diff --git a/backend/src/api/routers/webhooks.py b/backend/src/api/routers/webhooks.py index a07ff32..9432ef4 100644 --- a/backend/src/api/routers/webhooks.py +++ b/backend/src/api/routers/webhooks.py @@ -61,13 +61,16 @@ async def list_webhooks( async def create_webhook( pool: FromDishka[asyncpg.Pool], body: WebhookCreate ) -> WebhookView: - return await webhooks.create_webhook( + webhook = await webhooks.create_webhook( pool, body.account_id, _url(body.url), _watch_ids(body.watch_ids), enabled=body.enabled, ) + if webhook is None: + raise HTTPException(status_code=500, detail="webhook was not saved") + return webhook @router.put("/{webhook_id}") diff --git a/backend/src/utils/read/models.py b/backend/src/utils/read/models.py index 29371eb..7ebc580 100644 --- a/backend/src/utils/read/models.py +++ b/backend/src/utils/read/models.py @@ -546,6 +546,8 @@ class PersonRef(BaseModel): class WatchItem(WatchView): chat: ChatRef | None person: PersonRef | None + alert_count: int + last_alert_at: datetime | None class AlertView(BaseModel): @@ -562,6 +564,7 @@ class AlertWatch(BaseModel): id: int kind: str name: str | None + params: dict[str, Any] class AlertMessage(BaseModel): @@ -583,6 +586,7 @@ class AlertItem(BaseModel): chat: ChatRef | None message: AlertMessage | None peer: PersonRef | None + away_since: datetime | None path: str @@ -592,17 +596,6 @@ class AlertPage(BaseModel): unseen: int -class WebhookView(BaseModel): - id: int - account_id: int - url: str - secret: str - enabled: bool - watch_ids: list[int] | None - created_at: datetime - updated_at: datetime - - class DeliveryView(BaseModel): id: int webhook_id: int @@ -617,6 +610,18 @@ class DeliveryView(BaseModel): finished_at: datetime | None +class WebhookView(BaseModel): + id: int + account_id: int + url: str + secret: str + enabled: bool + watch_ids: list[int] | None + created_at: datetime + updated_at: datetime + last_delivery: DeliveryView | None + + class FileShareView(BaseModel): id: int account_id: int diff --git a/backend/src/utils/read/watches.py b/backend/src/utils/read/watches.py index a7636e8..7f946dc 100644 --- a/backend/src/utils/read/watches.py +++ b/backend/src/utils/read/watches.py @@ -44,9 +44,15 @@ WITH alert AS ( SELECT * FROM alert """ # noqa: S608 +_WATCH_STATS = """ +SELECT watch_id, count(*) AS count, max(ts) AS last FROM alerts +WHERE watch_id = ANY($1::int[]) GROUP BY watch_id +""" + _ALERT_ROWS = """ SELECT a.id, a.account_id, a.ts, a.seen, a.payload, w.id AS watch_id, w.kind AS watch_kind, w.name AS watch_name, + w.params AS watch_params, m.text AS stored_text, m.sender_id AS stored_sender, m.date AS stored_date, m.deleted_at IS NOT NULL AS stored_deleted, @@ -225,9 +231,17 @@ async def describe_watches( {chat for chat, _ in targets if chat is not None}, {person for _, person in targets if person is not None}, ) + stats = { + row["watch_id"]: row + for row in await pool.fetch(_WATCH_STATS, [watch.id for watch in watches]) + } return [ WatchItem( - **watch.model_dump(), chat=refs.chat(chat), person=refs.person(person) + **watch.model_dump(), + chat=refs.chat(chat), + person=refs.person(person), + alert_count=stats[watch.id]["count"] if watch.id in stats else 0, + last_alert_at=stats[watch.id]["last"] if watch.id in stats else None, ) for watch, (chat, person) in zip(watches, targets, strict=True) ] @@ -267,11 +281,15 @@ def _alert_item(row: asyncpg.Record, refs: Refs) -> AlertItem: seen=row["seen"], event=event, watch=AlertWatch( - id=row["watch_id"], kind=row["watch_kind"], name=row["watch_name"] + id=row["watch_id"], + kind=row["watch_kind"], + name=row["watch_name"], + params=json.loads(row["watch_params"]), ), chat=refs.chat(chat_id), message=message, peer=refs.person(peer_id), + away_since=payload.get("since"), path=_path(chat_id, message_id, peer_id), ) diff --git a/backend/src/utils/read/webhooks.py b/backend/src/utils/read/webhooks.py index aeaf35c..5c05713 100644 --- a/backend/src/utils/read/webhooks.py +++ b/backend/src/utils/read/webhooks.py @@ -1,3 +1,4 @@ +import json import secrets from datetime import datetime @@ -11,12 +12,21 @@ STATUS_DELIVERED = "delivered" STATUS_FAILED = "failed" DELIVERY_LOG = 20 -_COLS = "id, account_id, url, secret, enabled, watch_ids, created_at, updated_at" _DELIVERY_COLS = ( "id, webhook_id, alert_id, event, status, attempts, next_attempt_at, " "response_code, last_error, created_at, finished_at" ) +_HOOKS = f""" +SELECT w.id, w.account_id, w.url, w.secret, w.enabled, w.watch_ids, + w.created_at, w.updated_at, to_jsonb(d) AS last_delivery +FROM webhooks w +LEFT JOIN LATERAL ( + SELECT {_DELIVERY_COLS} FROM webhook_deliveries + WHERE webhook_id = w.id ORDER BY id DESC LIMIT 1 +) d ON true +""" # noqa: S608 + _CLAIM = """ UPDATE webhook_deliveries d SET status = 'sending', attempts = d.attempts + 1 FROM webhooks w @@ -43,20 +53,24 @@ def new_secret() -> str: return secrets.token_hex(32) +def _to_webhook(row: asyncpg.Record) -> WebhookView: + data = dict(row) + last = data.pop("last_delivery") + return WebhookView( + **data, last_delivery=DeliveryView(**json.loads(last)) if last else None + ) + + async def list_webhooks(pool: asyncpg.Pool, account_id: int) -> list[WebhookView]: rows = await pool.fetch( - f"SELECT {_COLS} FROM webhooks WHERE account_id = $1 ORDER BY id", # noqa: S608 - account_id, + f"{_HOOKS} WHERE w.account_id = $1 ORDER BY w.id", account_id ) - return [WebhookView(**dict(row)) for row in rows] + return [_to_webhook(row) for row in rows] async def get_webhook(pool: asyncpg.Pool, webhook_id: int) -> WebhookView | None: - row = await pool.fetchrow( - f"SELECT {_COLS} FROM webhooks WHERE id = $1", # noqa: S608 - webhook_id, - ) - return WebhookView(**dict(row)) if row else None + row = await pool.fetchrow(f"{_HOOKS} WHERE w.id = $1", webhook_id) + return _to_webhook(row) if row else None async def create_webhook( @@ -66,17 +80,17 @@ async def create_webhook( watch_ids: list[int] | None, *, enabled: bool, -) -> WebhookView: - row = await pool.fetchrow( - "INSERT INTO webhooks (account_id, url, secret, enabled, watch_ids) " # noqa: S608 - f"VALUES ($1, $2, $3, $4, $5) RETURNING {_COLS}", +) -> WebhookView | None: + webhook_id = await pool.fetchval( + "INSERT INTO webhooks (account_id, url, secret, enabled, watch_ids) " + "VALUES ($1, $2, $3, $4, $5) RETURNING id", account_id, url, new_secret(), enabled, watch_ids, ) - return WebhookView(**dict(row)) + return await get_webhook(pool, webhook_id) async def update_webhook( @@ -88,32 +102,31 @@ async def update_webhook( enabled: bool, ) -> WebhookView | None: async with pool.acquire() as conn, conn.transaction(): - row = await conn.fetchrow( - "UPDATE webhooks SET url = $2, enabled = $3, watch_ids = $4, " # noqa: S608 - f"updated_at = now() WHERE id = $1 RETURNING {_COLS}", + found = await conn.fetchval( + "UPDATE webhooks SET url = $2, enabled = $3, watch_ids = $4, " + "updated_at = now() WHERE id = $1 RETURNING id", webhook_id, url, enabled, watch_ids, ) - if row is not None and not enabled: + if found is not None and not enabled: await conn.execute( "UPDATE webhook_deliveries SET status = 'failed', " "last_error = 'endpoint disabled', finished_at = now() " "WHERE webhook_id = $1 AND status = 'pending'", webhook_id, ) - return WebhookView(**dict(row)) if row else None + return await get_webhook(pool, webhook_id) async def rotate_secret(pool: asyncpg.Pool, webhook_id: int) -> WebhookView | None: - row = await pool.fetchrow( - "UPDATE webhooks SET secret = $2, updated_at = now() " # noqa: S608 - f"WHERE id = $1 RETURNING {_COLS}", + await pool.execute( + "UPDATE webhooks SET secret = $2, updated_at = now() WHERE id = $1", webhook_id, new_secret(), ) - return WebhookView(**dict(row)) if row else None + return await get_webhook(pool, webhook_id) async def delete_webhook(pool: asyncpg.Pool, webhook_id: int) -> bool: