refactor(core): reuse dispatch idempotency for event tasks

Author: dcramer

…unior/migrations/0014_fuzzy_toad_men.sql …r/migrations/0014_normal_kat_farrell.sqlpackages/junior/migrations/0014_fuzzy_toad_men.sql renamed to packages/junior/migrations/0014_normal_kat_farrell.sql packages/junior/migrations/0014_fuzzy_toad_men.sql renamed to packages/junior/migrations/0014_normal_kat_farrell.sql

@@ -1,11 +1,3 @@
-CREATE TABLE "junior_event_task_matches" (
-	"id" text PRIMARY KEY NOT NULL,
-	"task_id" text NOT NULL,
-	"status" text NOT NULL,
-	"created_at_ms" bigint NOT NULL,
-	"match_json" jsonb NOT NULL
-);
---> statement-breakpoint
 CREATE TABLE "junior_event_tasks" (
 	"id" text PRIMARY KEY NOT NULL,
 	"team_id" text NOT NULL,
@@ -16,6 +8,5 @@ CREATE TABLE "junior_event_tasks" (
 	"task_json" jsonb NOT NULL
 );
 --> statement-breakpoint
-CREATE INDEX "junior_event_task_matches_task_idx" ON "junior_event_task_matches" USING btree ("task_id","created_at_ms","id");--> statement-breakpoint
 CREATE INDEX "junior_event_tasks_team_idx" ON "junior_event_tasks" USING btree ("team_id","created_at_ms","id") WHERE "junior_event_tasks"."status" <> 'deleted';--> statement-breakpoint
 CREATE INDEX "junior_event_tasks_match_idx" ON "junior_event_tasks" USING btree ("provider","resource_ref","created_at_ms","id") WHERE "junior_event_tasks"."status" = 'active';

packages/junior/migrations/meta/0014_snapshot.json

@@ -1,5 +1,5 @@
 {
-  "id": "49fab1ec-994f-469b-9eb9-94c0d737bb9d",
+  "id": "2d78fbea-150b-420d-bbc7-7fa4927211e0",
   "prevId": "5d0244f5-7ca6-4a03-ad59-4699ba6bf0e1",
   "version": "7",
   "dialect": "postgresql",
@@ -155,25 +155,16 @@
           "name": "junior_conversation_annotations_conversation_id_fk",
           "tableFrom": "junior_conversation_annotations",
           "tableTo": "junior_conversations",
-          "columnsFrom": [
-            "conversation_id"
-          ],
-          "columnsTo": [
-            "conversation_id"
-          ],
+          "columnsFrom": ["conversation_id"],
+          "columnsTo": ["conversation_id"],
           "onDelete": "cascade",
           "onUpdate": "no action"
         }
       },
       "compositePrimaryKeys": {
         "junior_conversation_annotations_pk": {
           "name": "junior_conversation_annotations_pk",
-          "columns": [
-            "conversation_id",
-            "plugin",
-            "kind",
-            "key"
-          ]
+          "columns": ["conversation_id", "plugin", "kind", "key"]
         }
       },
       "uniqueConstraints": {},
@@ -333,23 +324,16 @@
           "name": "junior_conversation_events_conversation_id_junior_conversations_conversation_id_fk",
           "tableFrom": "junior_conversation_events",
           "tableTo": "junior_conversations",
-          "columnsFrom": [
-            "conversation_id"
-          ],
-          "columnsTo": [
-            "conversation_id"
-          ],
+          "columnsFrom": ["conversation_id"],
+          "columnsTo": ["conversation_id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         }
       },
       "compositePrimaryKeys": {
         "junior_conversation_events_conversation_id_seq_pk": {
           "name": "junior_conversation_events_conversation_id_seq_pk",
-          "columns": [
-            "conversation_id",
-            "seq"
-          ]
+          "columns": ["conversation_id", "seq"]
         }
       },
       "uniqueConstraints": {},
@@ -706,77 +690,53 @@
           "name": "junior_conversations_destination_id_junior_destinations_id_fk",
           "tableFrom": "junior_conversations",
           "tableTo": "junior_destinations",
-          "columnsFrom": [
-            "destination_id"
-          ],
-          "columnsTo": [
-            "id"
-          ],
+          "columnsFrom": ["destination_id"],
+          "columnsTo": ["id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         },
         "junior_conversations_actor_identity_id_junior_identities_id_fk": {
           "name": "junior_conversations_actor_identity_id_junior_identities_id_fk",
           "tableFrom": "junior_conversations",
           "tableTo": "junior_identities",
-          "columnsFrom": [
-            "actor_identity_id"
-          ],
-          "columnsTo": [
-            "id"
-          ],
+          "columnsFrom": ["actor_identity_id"],
+          "columnsTo": ["id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         },
         "junior_conversations_creator_identity_id_junior_identities_id_fk": {
           "name": "junior_conversations_creator_identity_id_junior_identities_id_fk",
           "tableFrom": "junior_conversations",
           "tableTo": "junior_identities",
-          "columnsFrom": [
-            "creator_identity_id"
-          ],
-          "columnsTo": [
-            "id"
-          ],
+          "columnsFrom": ["creator_identity_id"],
+          "columnsTo": ["id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         },
         "junior_conversations_credential_subject_identity_id_junior_identities_id_fk": {
           "name": "junior_conversations_credential_subject_identity_id_junior_identities_id_fk",
           "tableFrom": "junior_conversations",
           "tableTo": "junior_identities",
-          "columnsFrom": [
-            "credential_subject_identity_id"
-          ],
-          "columnsTo": [
-            "id"
-          ],
+          "columnsFrom": ["credential_subject_identity_id"],
+          "columnsTo": ["id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         },
         "junior_conversations_parent_conversation_id_junior_conversations_conversation_id_fk": {
           "name": "junior_conversations_parent_conversation_id_junior_conversations_conversation_id_fk",
           "tableFrom": "junior_conversations",
           "tableTo": "junior_conversations",
-          "columnsFrom": [
-            "parent_conversation_id"
-          ],
-          "columnsTo": [
-            "conversation_id"
-          ],
+          "columnsFrom": ["parent_conversation_id"],
+          "columnsTo": ["conversation_id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         },
         "junior_conversations_root_conversation_id_junior_conversations_conversation_id_fk": {
           "name": "junior_conversations_root_conversation_id_junior_conversations_conversation_id_fk",
           "tableFrom": "junior_conversations",
           "tableTo": "junior_conversations",
-          "columnsFrom": [
-            "root_conversation_id"
-          ],
-          "columnsTo": [
-            "conversation_id"
-          ],
+          "columnsFrom": ["root_conversation_id"],
+          "columnsTo": ["conversation_id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         }
@@ -917,77 +877,6 @@
       "checkConstraints": {},
       "isRLSEnabled": false
     },
-    "public.junior_event_task_matches": {
-      "name": "junior_event_task_matches",
-      "schema": "",
-      "columns": {
-        "id": {
-          "name": "id",
-          "type": "text",
-          "primaryKey": true,
-          "notNull": true
-        },
-        "task_id": {
-          "name": "task_id",
-          "type": "text",
-          "primaryKey": false,
-          "notNull": true
-        },
-        "status": {
-          "name": "status",
-          "type": "text",
-          "primaryKey": false,
-          "notNull": true
-        },
-        "created_at_ms": {
-          "name": "created_at_ms",
-          "type": "bigint",
-          "primaryKey": false,
-          "notNull": true
-        },
-        "match_json": {
-          "name": "match_json",
-          "type": "jsonb",
-          "primaryKey": false,
-          "notNull": true
-        }
-      },
-      "indexes": {
-        "junior_event_task_matches_task_idx": {
-          "name": "junior_event_task_matches_task_idx",
-          "columns": [
-            {
-              "expression": "task_id",
-              "isExpression": false,
-              "asc": true,
-              "nulls": "last"
-            },
-            {
-              "expression": "created_at_ms",
-              "isExpression": false,
-              "asc": true,
-              "nulls": "last"
-            },
-            {
-              "expression": "id",
-              "isExpression": false,
-              "asc": true,
-              "nulls": "last"
-            }
-          ],
-          "isUnique": false,
-          "concurrently": false,
-          "method": "btree",
-          "with": {}
-        }
-      },
-      "foreignKeys": {},
-      "compositePrimaryKeys": {},
-      "uniqueConstraints": {},
-      "policies": {},
-      "checkConstraints": {},
-      "isRLSEnabled": false
-    },
     "public.junior_event_tasks": {
       "name": "junior_event_tasks",
       "schema": "",
@@ -1289,12 +1178,8 @@
           "name": "junior_identities_user_id_junior_users_id_fk",
           "tableFrom": "junior_identities",
           "tableTo": "junior_users",
-          "columnsFrom": [
-            "user_id"
-          ],
-          "columnsTo": [
-            "id"
-          ],
+          "columnsFrom": ["user_id"],
+          "columnsTo": ["id"],
           "onDelete": "no action",
           "onUpdate": "no action"
         }
@@ -1346,12 +1231,7 @@
       "compositePrimaryKeys": {
         "junior_stats_date_namespace_metric_name_pk": {
           "name": "junior_stats_date_namespace_metric_name_pk",
-          "columns": [
-            "date",
-            "namespace",
-            "metric",
-            "name"
-          ]
+          "columns": ["date", "namespace", "metric", "name"]
         }
       },
       "uniqueConstraints": {},
@@ -1436,4 +1316,4 @@
     "schemas": {},
     "tables": {}
   }
-}
+}

packages/junior/migrations/meta/_journal.json

@@ -103,9 +103,9 @@
     {
       "idx": 14,
       "version": "7",
-      "when": 1785525804752,
-      "tag": "0014_fuzzy_toad_men",
+      "when": 1785546394600,
+      "tag": "0014_normal_kat_farrell",
       "breakpoints": true
     }
   ]
-}
+}

packages/junior/src/chat/event-tasks/README.md

@@ -1,11 +1,14 @@
 # Event tasks
 
 Core stores durable Slack-destination tasks that match normalized provider
-resource events by provider, resource ref, and event type. Provider plugins own webhook
-verification and event normalization; they publish through the core resource
-event boundary without knowing which conversations or tasks consume an event.
+resource events by provider, resource ref, and event type. Provider plugins own
+webhook verification and event normalization; they publish through the core
+resource event boundary without knowing which conversations or tasks consume an
+event.
 
 Each matching task receives an independent idempotent agent dispatch. A failure
 for one task does not prevent dispatch attempts for other matching tasks; the
 ingress boundary still receives the aggregate failure for provider retry. Task
 management is bound to the Slack destination where the task was created.
+One Junior deployment serves one Slack workspace, which is the event-task
+authorization boundary.

packages/junior/src/chat/event-tasks/ingest.ts

@@ -5,17 +5,17 @@ import {
   type ResourceEvent,
 } from "@sentry/junior-plugin-api";
 import { dispatchTask } from "@/chat/agent-dispatch/context";
-import { getDb, getSqlExecutor } from "@/chat/db";
-import {
-  claimActiveEventTaskMatch,
-  findMatchingEventTasks,
-  markEventTaskMatchDispatched,
-} from "@/chat/event-tasks/store";
-import type { EventTask, EventTaskMatch } from "@/chat/event-tasks/types";
+import { getDb } from "@/chat/db";
+import { findMatchingEventTasks } from "@/chat/event-tasks/store";
+import type { EventTask } from "@/chat/event-tasks/types";
 import type { ConversationWorkQueue } from "@/chat/task-execution/queue";
 
-function matchId(taskId: string, provider: string, eventKey: string): string {
-  return `evmatch_${createHash("sha256")
+function eventTaskDispatchKey(
+  taskId: string,
+  provider: string,
+  eventKey: string,
+): string {
+  return `event-task:${createHash("sha256")
     .update(`${taskId}\0${provider}\0${eventKey}`)
     .digest("hex")
     .slice(0, 32)}`;
@@ -68,18 +68,11 @@ export async function ingestEventTasks(
   const errors: unknown[] = [];
   for (const task of tasks) {
     try {
-      const id = matchId(task.id, event.provider, event.eventKey);
-      const pending: EventTaskMatch = {
-        id,
-        createdAtMs: nowMs,
-        eventKey: event.eventKey,
-        eventType: event.eventType,
-        status: "pending",
-        taskId: task.id,
-      };
-      const match = await claimActiveEventTaskMatch(getSqlExecutor(), pending);
-      if (!match) continue;
-      if (match.status === "dispatched") continue;
+      const idempotencyKey = eventTaskDispatchKey(
+        task.id,
+        event.provider,
+        event.eventKey,
+      );
       const credentialSubject =
         task.credentialMode === "creator"
           ? {
@@ -93,7 +86,7 @@ export async function ingestEventTasks(
         conversationWorkQueue: options.queue,
         nowMs,
         options: {
-          idempotencyKey: id,
+          idempotencyKey,
           ...(credentialSubject ? { credentialSubject } : {}),
           destination: task.destination,
           destinationVisibility:
@@ -116,8 +109,9 @@ export async function ingestEventTasks(
           }),
         },
       });
-      await markEventTaskMatchDispatched(db, id, dispatch.id);
-      dispatched += 1;
+      if (dispatch.status === "created") {
+        dispatched += 1;
+      }
     } catch (error) {
       errors.push(error);
     }