Fire automation events across API routers
FlowDeck CI / lint (push) Failing after 1m12s
FlowDeck CI / test (push) Failing after 3h3m3s
FlowDeck CI / docker (push) Skipped

This commit is contained in:
2026-09-21 20:30:05 -04:00
parent 189ed5bbca
commit e0237e576f
8 changed files with 202 additions and 2 deletions
+64
View File
@@ -16,6 +16,7 @@ from fastapi import APIRouter, Header, HTTPException, Request
from fastapi.responses import JSONResponse, Response
from app.db import get_conn
from app.services.automations import fire_event as _fire_event
from app.services.api_v2_helpers import (
audit_log,
check_idempotency,
@@ -779,6 +780,10 @@ async def create_collection_page_v2(collection_id: int, request: Request, author
conn.commit()
row = conn.execute("SELECT * FROM collection_pages WHERE id=?", (pid,)).fetchone()
audit_log(user, "page.create", "collection_page", pid, title, request)
try:
await _fire_event("collection.page.created", {"page_id": pid, "collection_id": collection_id, "title": title})
except Exception:
pass
data = {"id": pid, "title": title, "status": "created", "page": row_to_dict(row)}
key = (request.headers.get("Idempotency-Key") or "").strip()
if key:
@@ -845,6 +850,10 @@ async def patch_page_v2(page_id: int, request: Request, authorization: str | Non
conn.execute("UPDATE collection_pages SET title=?, icon=?, position=?, parent_id=?, property_values_json=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (title, icon, pos, parent_id, json.dumps(stored), page_id))
conn.commit()
audit_log(user, "page.update", "collection_page", page_id, "", request)
try:
await _fire_event("collection.page.updated", {"page_id": page_id, "collection_id": row["collection_id"], "title": title})
except Exception:
pass
return {"id": page_id, "status": "updated"}
@router.delete("/pages/{page_id}")
@@ -860,6 +869,10 @@ async def delete_page_v2(page_id: int, request: Request, authorization: str | No
conn.execute("DELETE FROM collection_pages WHERE id=?", (page_id,))
conn.commit()
audit_log(user, "page.delete", "collection_page", page_id, "", request)
try:
await _fire_event("collection.page.deleted", {"page_id": page_id, "collection_id": row["collection_id"]})
except Exception:
pass
return {"id": page_id, "status": "deleted"}
@router.post("/pages/{page_id}/restore")
@@ -874,6 +887,10 @@ async def restore_page_v2(page_id: int, request: Request, authorization: str | N
if row and row["deleted_at"]:
conn.execute("UPDATE pages SET deleted_at=NULL WHERE id=?", (page_id,))
conn.commit()
try:
await _fire_event("page.restored", {"page_id": page_id})
except Exception:
pass
return {"id": page_id, "status": "restored"}
raise HTTPException(404, "Page not found or not deleted")
@@ -1153,6 +1170,10 @@ async def create_view_v2(collection_id: int, request: Request, authorization: st
vid = cur.lastrowid
conn.commit()
audit_log(user, "view.create", "view", vid, name, request)
try:
await _fire_event("collection.view.created", {"view_id": vid, "collection_id": collection_id, "name": name, "view_type": vtype})
except Exception:
pass
return {"id": vid, "name": name, "view_type": vtype, "status": "created"}
@router.patch("/views/{view_id}")
@@ -1315,6 +1336,10 @@ async def create_comment_v2(page_id: int, request: Request, authorization: str |
conn.commit()
row = conn.execute("SELECT * FROM comments WHERE id=?", (nid,)).fetchone()
audit_log(user, "comment.create", "comment", nid, text[:80], request)
try:
await _fire_event("comment.added", {"comment_id": nid, "page_id": page_id, "user_id": user["id"]})
except Exception:
pass
return {"id": nid, "status": "created", "comment": row_to_dict(row)}
@router.patch("/comments/{comment_id}")
@@ -1335,8 +1360,14 @@ async def patch_comment_v2(comment_id: int, request: Request, authorization: str
raise HTTPException(403, "Not your comment")
body_text = body.get("body", row["body"])
resolved = body.get("resolved", row["resolved"])
was_resolved = int(row["resolved"] or 0)
conn.execute("UPDATE comments SET body=?, resolved=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (body_text, int(bool(resolved)), comment_id))
conn.commit()
if int(bool(resolved)) and not was_resolved:
try:
await _fire_event("comment.resolved", {"comment_id": comment_id, "page_id": row["page_id"]})
except Exception:
pass
return {"id": comment_id, "status": "updated"}
@router.delete("/comments/{comment_id}")
@@ -1379,6 +1410,11 @@ async def create_mention_v2(page_id: int, request: Request, authorization: str |
except Exception:
pass
conn.commit()
if created:
try:
await _fire_event("mention.added", {"page_id": page_id, "user_ids": [u for u in targets if isinstance(u, int)], "count": created})
except Exception:
pass
return {"mentions": created, "status": "created"}
# ── Notifications ─────────────────────────────────────────────────────────
@@ -1472,6 +1508,10 @@ async def add_favorite_v2(request: Request, authorization: str | None = Header(d
conn.commit()
except Exception as e:
raise HTTPException(409, str(e)) from None
try:
await _fire_event("favorite.added", {"page_id": pid, "user_id": user["id"]})
except Exception:
pass
return {"page_id": pid, "status": "added"}
@router.delete("/favorites/{page_id}")
@@ -1483,6 +1523,10 @@ async def remove_favorite_v2(page_id: int, request: Request, authorization: str
with get_conn() as conn:
conn.execute("DELETE FROM favorites WHERE user_id=? AND page_id=?", (user["id"], page_id))
conn.commit()
try:
await _fire_event("favorite.removed", {"page_id": page_id, "user_id": user["id"]})
except Exception:
pass
return {"page_id": page_id, "status": "removed"}
@router.get("/tags")
@@ -1629,6 +1673,10 @@ async def create_share_v2(page_id: int, request: Request, authorization: str | N
conn.commit()
nid = cur.lastrowid
audit_log(user, "share.create", "share", nid, f"page={page_id}", request)
try:
await _fire_event("page.shared", {"page_id": page_id, "share_id": nid, "permission": perm})
except Exception:
pass
return {"id": nid, "page_id": page_id, "status": "shared"}
@router.patch("/shares/{share_id}")
@@ -1686,6 +1734,10 @@ async def publish_page_v2(page_id: int, request: Request, authorization: str | N
conn.execute("UPDATE pages SET is_published=1, publish_slug=?, is_shared=1 WHERE id=?", (slug, page_id))
conn.commit()
audit_log(user, "page.publish", "page", page_id, slug, request)
try:
await _fire_event("page.published", {"page_id": page_id, "slug": slug})
except Exception:
pass
return {"page_id": page_id, "slug": slug, "url": f"/p/{slug}", "status": "published"}
@router.delete("/pages/{page_id}/publish")
@@ -1697,6 +1749,10 @@ async def unpublish_page_v2(page_id: int, request: Request, authorization: str |
with get_conn() as conn:
conn.execute("UPDATE pages SET is_published=0 WHERE id=?", (page_id,))
conn.commit()
try:
await _fire_event("page.unpublished", {"page_id": page_id})
except Exception:
pass
return {"page_id": page_id, "status": "unpublished"}
# ── History ───────────────────────────────────────────────────────────────
@@ -1774,6 +1830,10 @@ async def create_sprint_v2(collection_id: int, request: Request, authorization:
cur = conn.execute("INSERT INTO sprints (collection_id, name, start_date, end_date, goal) VALUES (?, ?, ?, ?, ?)", (collection_id, name, start, end, body.get("goal") or ""))
sid = cur.lastrowid
conn.commit()
try:
await _fire_event("sprint.created", {"sprint_id": sid, "collection_id": collection_id, "name": name})
except Exception:
pass
return {"id": sid, "name": name, "status": "created"}
@router.patch("/sprints/{sprint_id}")
@@ -1797,6 +1857,10 @@ async def patch_sprint_v2(sprint_id: int, request: Request, authorization: str |
status = body.get("status", row["status"])
conn.execute("UPDATE sprints SET name=?, start_date=?, end_date=?, goal=?, status=? WHERE id=?", (name, start, end, goal, status, sprint_id))
conn.commit()
try:
await _fire_event("sprint.updated", {"sprint_id": sprint_id, "collection_id": row["collection_id"], "name": name, "status": status})
except Exception:
pass
return {"id": sprint_id, "status": "updated"}
@router.delete("/sprints/{sprint_id}")
+20
View File
@@ -978,6 +978,10 @@ async def add_favorite(request: Request, page_id: int):
(uid, page_id, pos),
)
conn.commit()
try:
await fire_event("favorite.added", {"page_id": page_id, "user_id": uid})
except Exception:
pass
return {"status": "added", "page_id": page_id}
@@ -989,6 +993,10 @@ async def remove_favorite(request: Request, page_id: int):
with get_conn() as conn:
conn.execute("DELETE FROM favorites WHERE user_id=? AND page_id=?", (uid, page_id))
conn.commit()
try:
await fire_event("favorite.removed", {"page_id": page_id, "user_id": uid})
except Exception:
pass
return {"status": "removed", "page_id": page_id}
# ═══════════ Share API ═══════════
@@ -1020,6 +1028,10 @@ async def publish_page(request: Request, page_id: int):
)
conn.commit()
row = conn.execute("SELECT title FROM pages WHERE id=?", (page_id,)).fetchone()
try:
await fire_event("page.published", {"page_id": page_id, "slug": slug})
except Exception:
pass
return {"is_published": True, "publish_slug": slug, "title": row["title"] if row else ""}
@@ -1032,6 +1044,10 @@ async def unpublish_page(request: Request, page_id: int):
(page_id,),
)
conn.commit()
try:
await fire_event("page.unpublished", {"page_id": page_id})
except Exception:
pass
return {"is_published": False}
@@ -1049,6 +1065,10 @@ async def restore_page(request: Request, page_id: int):
with get_conn() as conn:
conn.execute("UPDATE pages SET deleted_at=NULL WHERE id=?", (page_id,))
conn.commit()
try:
await fire_event("page.restored", {"page_id": page_id})
except Exception:
pass
return {"status": "ok", "restored": page_id}
+20
View File
@@ -13,6 +13,7 @@ from fastapi import APIRouter, HTTPException, Request
from app.auth.session import SessionManager
from app.db import get_conn
from app.services import notifications as notif
from app.services.automations import fire_event as _fire_event
logger = logging.getLogger(__name__)
router = APIRouter(tags=["collaboration"], prefix="/api")
@@ -116,6 +117,14 @@ async def add_comment(request: Request, page_id: int):
)
conn.commit()
try:
await _fire_event("comment.added", {"comment_id": comment_id, "page_id": page_id, "user_id": uid})
mentioned_ids = notif.extract_mentions(text)
if mentioned_ids:
await _fire_event("mention.added", {"page_id": page_id, "comment_id": comment_id, "count": len(mentioned_ids)})
except Exception:
pass
return {"id": comment_id, "status": "created"}
@@ -140,6 +149,11 @@ async def notify_page_mentions(request: Request, page_id: int):
"page", page_id, url, conn=conn,
)
conn.commit()
if mentioned:
try:
await _fire_event("mention.added", {"page_id": page_id, "user_ids": mentioned, "count": len(mentioned)})
except Exception:
pass
return {"mentioned": mentioned}
@@ -161,10 +175,16 @@ async def update_comment(request: Request, comment_id: int):
"UPDATE comments SET body=?, updated_at=CURRENT_TIMESTAMP WHERE id=?",
(body["body"].strip(), comment_id),
)
was_resolved = int(row["resolved"] or 0)
if "resolved" in body and body.get("resolved") is not None:
conn.execute("UPDATE comments SET resolved=? WHERE id=?",
(1 if body["resolved"] else 0, comment_id))
conn.commit()
if body.get("resolved") and not was_resolved:
try:
await _fire_event("comment.resolved", {"comment_id": comment_id, "page_id": row["page_id"]})
except Exception:
pass
return {"id": comment_id, "status": "updated"}
+35 -2
View File
@@ -561,6 +561,11 @@ async def update_page_api(request: Request, page_id: int):
"icon": icon,
"properties": props,
})
await fire_event("collection.page.updated", {
"page_id": page_id,
"collection_id": existing["collection_id"],
"title": title,
})
# Notify newly assigned people (person properties) — v5.8.0.
from app.services.notifications import notify_assignment
user = _current_user(request)
@@ -589,6 +594,10 @@ async def delete_page_api(request: Request, page_id: int):
"collection_id": existing["collection_id"],
"title": existing["title"],
})
await fire_event("collection.page.deleted", {
"page_id": page_id,
"collection_id": existing["collection_id"],
})
return {"id": page_id, "status": "deleted"}
@@ -1185,8 +1194,15 @@ async def save_view_as(request: Request, collection_id: int):
(collection_id, name, view_type, json.dumps(config), max_pos, user_id),
)
conn.commit()
new_view_id = cur.lastrowid
return {"id": cur.lastrowid, "name": name, "view_type": view_type,
await fire_event("collection.view.created", {
"view_id": new_view_id,
"collection_id": collection_id,
"name": name,
"view_type": view_type,
})
return {"id": new_view_id, "name": name, "view_type": view_type,
"config_json": json.dumps(config), "created_by": user_id, "status": "saved"}
@@ -1256,8 +1272,15 @@ async def duplicate_view_api(request: Request, view_id: int):
existing["config_json"], max_pos, user_id),
)
conn.commit()
dup_view_id = cur.lastrowid
return {"id": cur.lastrowid, "name": name, "view_type": existing["view_type"],
await fire_event("collection.view.created", {
"view_id": dup_view_id,
"collection_id": existing["collection_id"],
"name": name,
"view_type": existing["view_type"],
})
return {"id": dup_view_id, "name": name, "view_type": existing["view_type"],
"status": "duplicated"}
@@ -1319,6 +1342,11 @@ async def create_sub_item(request: Request, collection_id: int, page_id: int):
"title": title,
"properties": body.get("properties", {}),
})
await fire_event("collection.page.created", {
"page_id": new_id,
"collection_id": collection_id,
"title": title,
})
return {"id": new_id, "title": title, "parent_id": page_id, "status": "created"}
@@ -2485,4 +2513,9 @@ async def create_page_api(request: Request, collection_id: int):
"icon": icon,
"properties": property_values,
})
await fire_event("collection.page.created", {
"page_id": page_id,
"collection_id": collection_id,
"title": title,
})
return {"id": page_id, "title": title, "status": "created"}
+16
View File
@@ -10,6 +10,7 @@ from fastapi import APIRouter, HTTPException, Request
from app.auth.session import SessionManager
from app.db import get_conn
from app.services.automations import fire_event as _fire_event
logger = logging.getLogger(__name__)
router = APIRouter(tags=["sharing"], prefix="/api")
@@ -126,6 +127,11 @@ async def share_page(page_id: int, request: Request):
logger.warning("share→page_permissions mirror failed (page=%s group=%s)", page_id, target_group_id)
conn.commit()
try:
await _fire_event("page.shared", {"page_id": page_id, "share_id": share_id, "permission": permission})
except Exception:
pass
return {
"id": share_id,
"page_id": page_id,
@@ -323,6 +329,11 @@ async def publish_page(page_id: int, request: Request):
)
conn.commit()
try:
await _fire_event("page.published", {"page_id": page_id, "slug": slug})
except Exception:
pass
return {
"page_id": page_id,
"is_published": True,
@@ -349,6 +360,11 @@ async def unpublish_page(page_id: int, request: Request):
)
conn.commit()
try:
await _fire_event("page.unpublished", {"page_id": page_id})
except Exception:
pass
return {
"page_id": page_id,
"is_published": False,
+22
View File
@@ -130,6 +130,10 @@ async def add_comment(request: Request, page_id: int):
cur = conn.execute("INSERT INTO comments (page_id, user_id, body, parent_id) VALUES (?,?,?,?)",
(page_id, uid, b, parent_id))
conn.commit()
try:
await fire_event("comment.added", {"comment_id": cur.lastrowid, "page_id": page_id, "user_id": uid})
except Exception:
pass
return {"id": cur.lastrowid, "status": "created"}
@@ -139,11 +143,17 @@ async def update_comment(request: Request, comment_id: int):
b = body.get("body")
resolved = body.get("resolved")
with get_conn() as conn:
row = conn.execute("SELECT page_id, resolved FROM comments WHERE id=?", (comment_id,)).fetchone()
if b is not None:
conn.execute("UPDATE comments SET body=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (b, comment_id))
if resolved is not None:
conn.execute("UPDATE comments SET resolved=? WHERE id=?", (int(resolved), comment_id))
conn.commit()
if resolved and row and not int(row["resolved"] or 0):
try:
await fire_event("comment.resolved", {"comment_id": comment_id, "page_id": row["page_id"]})
except Exception:
pass
return {"status": "updated"}
@@ -206,6 +216,10 @@ async def add_favorite(request: Request):
(uid, page_id, collection_id),
)
conn.commit()
try:
await fire_event("favorite.added", {"page_id": page_id, "collection_id": collection_id, "user_id": uid})
except Exception:
pass
return {"status": "favorited"}
@@ -449,6 +463,10 @@ async def create_sprint(request: Request, collection_id: int):
(collection_id, name, start_date, end_date, goal, status, auto_complete),
)
conn.commit()
try:
await fire_event("sprint.created", {"sprint_id": cur.lastrowid, "collection_id": collection_id, "name": name})
except Exception:
pass
return {"id": cur.lastrowid, "name": name, "status": "created"}
@@ -475,6 +493,10 @@ async def update_sprint(request: Request, collection_id: int, sid: int):
(name, start_date, end_date, goal, status, auto_complete, sid),
)
conn.commit()
try:
await fire_event("sprint.updated", {"sprint_id": sid, "collection_id": collection_id, "name": name, "status": status})
except Exception:
pass
return {"id": sid, "status": "updated"}
+11
View File
@@ -259,6 +259,17 @@ class AgentEngine:
self._persist_message(conversation_id, "assistant", final_text,
model=used_model, tokens=self._tokens)
await self._autotitle(conversation_id, objective, final_text)
# v6.4.0: emit agent.run.finished (outbound webhooks only).
try:
from app.services.webhook_outbound import fire_event as _fire_wh
await _fire_wh("agent.run.finished", {
"conversation_id": conversation_id,
"objective": objective[:500],
"model": used_model,
"tokens": self._tokens,
})
except Exception: # noqa: BLE001
logger.debug("agent.run.finished webhook dispatch failed")
except Exception as exc: # noqa: BLE001
logger.exception("AgentEngine run failed")
+14
View File
@@ -283,6 +283,20 @@ async def run_automation(automation_id: int, trigger_source: str, context: dict)
detail = "; ".join(results)
_save_run(automation_id, trigger_source, "fired", detail,
ctx.get("collection_id"), ctx.get("page_id"))
# v6.4.0: emit automation.fired (goes through fire_event → outbound
# webhooks, but NOT back through automations to avoid recursion).
try:
from app.services.webhook_outbound import fire_event as _fire_wh
await _fire_wh("automation.fired", {
"automation_id": automation_id,
"name": auto["name"],
"trigger": trigger_source,
"collection_id": ctx.get("collection_id"),
"page_id": ctx.get("page_id"),
"detail": detail,
})
except Exception: # noqa: BLE001
logger.debug("automation.fired webhook dispatch failed")
return {"status": "fired", "detail": detail}
except Exception as exc: # noqa: BLE001 — record every failure in history
logger.warning("Automation %s failed: %s", automation_id, exc)