Skip to content

Commit 2ed5d52

Browse files
kathiekiwikathiekiwi
committed
refactor(dashboard-agent-db): move the watch tables into their own schema module
Pure move, re-exported from schema.ts so drizzle-kit reads the same six tables and the generated SQL is unchanged.
1 parent 30ec754 commit 2ed5d52

4 files changed

Lines changed: 145 additions & 125 deletions

File tree

‎internal-packages/dashboard-agent-db/src/schema-base.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
import { pgSchema } from "drizzle-orm/pg-core";
2+
3+
// Tables are schema-qualified explicitly, so the connection needs no `search_path`.
4+
export const dashboardAgentSchema = pgSchema("trigger_dashboard_agent");

‎internal-packages/dashboard-agent-db/src/schema.ts‎

Lines changed: 5 additions & 123 deletions
Original file line numberDiff line numberDiff line change
@@ -4,21 +4,17 @@ import {
44
index,
55
integer,
66
jsonb,
7-
pgSchema,
87
primaryKey,
98
smallint,
109
text,
1110
timestamp,
12-
uniqueIndex,
1311
} from "drizzle-orm/pg-core";
14-
import type {
15-
WatchObservedOutcome,
16-
WatchResolution,
17-
WatchSpec,
18-
} from "@internal/dashboard-agent-contracts";
12+
import { dashboardAgentSchema } from "./schema-base.js";
1913

20-
// Tables are schema-qualified explicitly, so the connection needs no `search_path`.
21-
export const dashboardAgentSchema = pgSchema("trigger_dashboard_agent");
14+
// drizzle-kit reads this file, so the watch tables are re-exported here: the
15+
// generated SQL and the public surface stay exactly as they were.
16+
export * from "./schema-base.js";
17+
export * from "./watch-schema.js";
2218

2319
/**
2420
* Scoped to org + user. `organizationId` and `userId` are main-DB ids with no FK:
@@ -143,117 +139,6 @@ export const investigations = dashboardAgentSchema.table(
143139
]
144140
);
145141

146-
/** `active` is the only non-terminal status; the other three are immutable. */
147-
export type WatchStatus = "active" | "fired" | "expired" | "cancelled";
148-
/**
149-
* `not_required` while active and for every cancellation. `delivering` is the
150-
* in-flight claim, one deliverer at a time. See `claimWatchDelivery`.
151-
*/
152-
export type WatchDeliveryStatus = "not_required" | "pending" | "delivering" | "delivered";
153-
/** Cancellations are silent: no resolution, no wake. */
154-
export type WatchCancelReason = "user" | "access_revoked" | "chat_deleted" | "scheduling_failed";
155-
156-
/** `since` is server-set at creation, so `error_recurrence` can't match older errors. */
157-
export type PersistedWatchSpec = WatchSpec & { since?: string };
158-
159-
/**
160-
* The initiating identity is snapshotted at creation, so a membership change can only
161-
* cancel a watch, never widen its scope. Main-DB ids, FK-free.
162-
*/
163-
export const watches = dashboardAgentSchema.table(
164-
"watches",
165-
{
166-
id: text("id").primaryKey(), // = watchId (`watch_…`)
167-
chatId: text("chat_id").notNull(), // = chats.id
168-
identity: text("identity").notNull(),
169-
spec: jsonb("spec").$type<PersistedWatchSpec>().notNull(),
170-
status: text("status").$type<WatchStatus>().notNull().default("active"),
171-
deliveryStatus: text("delivery_status")
172-
.$type<WatchDeliveryStatus>()
173-
.notNull()
174-
.default("not_required"),
175-
cancelReason: text("cancel_reason").$type<WatchCancelReason>(),
176-
/**
177-
* The meaning; `status` above stays the two-value transport encoding so persisted
178-
* wake ids and dedup keys remain valid. NULL while active and on cancellation.
179-
*/
180-
resolution: text("resolution").$type<WatchResolution>(),
181-
/** Written in the same statement as `resolution` and `lastResult`. */
182-
observedOutcome: jsonb("observed_outcome").$type<WatchObservedOutcome>(),
183-
/** Consent given at creation. Never part of `identity`. */
184-
investigateOnAttention: boolean("investigate_on_attention").notNull().default(false),
185-
// Immutable initiating identity, snapshotted at creation.
186-
organizationId: text("organization_id").notNull(),
187-
projectId: text("project_id").notNull(),
188-
/** Nullable: rows created before this column don't carry it. */
189-
projectRef: text("project_ref"),
190-
environmentId: text("environment_id").notNull(),
191-
userId: text("user_id").notNull(),
192-
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
193-
expiresAt: timestamp("expires_at", { withTimezone: true }).notNull(),
194-
lastCheckedAt: timestamp("last_checked_at", { withTimezone: true }),
195-
firedAt: timestamp("fired_at", { withTimezone: true }),
196-
// A claim older than the delivery grace is abandoned and may be re-claimed.
197-
deliveryClaimedAt: timestamp("delivery_claimed_at", { withTimezone: true }),
198-
// Fencing token, written fresh on every claim and required by the release and
199-
// delivered marks, so a revived deliverer can't touch the claim that replaced it.
200-
deliveryClaimId: text("delivery_claim_id"),
201-
deliveredAt: timestamp("delivered_at", { withTimezone: true }),
202-
cancelledAt: timestamp("cancelled_at", { withTimezone: true }),
203-
// On fire or expire this is the notification's payload.
204-
lastResult: jsonb("last_result").$type<Record<string, unknown>>(),
205-
// Check idempotency keys are `watch:{id}:tick:{n}`.
206-
tickCount: integer("tick_count").notNull().default(0),
207-
},
208-
(t) => [
209-
index("watches_chat_idx").on(t.chatId),
210-
// UNIQUE is the dedup guarantee: a read-then-insert check can't be race-proof
211-
// under READ COMMITTED. Leading `chat_id` also serves the active-watches lookup.
212-
uniqueIndex("watches_chat_active_identity_key")
213-
.on(t.chatId, t.projectId, t.environmentId, t.identity)
214-
.where(sql`${t.status} = 'active'`),
215-
// Sweep: active watches due to be checked / past their expiry.
216-
index("watches_status_expires_idx").on(t.status, t.expiresAt),
217-
// Sweep: resolved watches whose wake is still owed.
218-
index("watches_pending_delivery_idx")
219-
.on(t.firedAt, t.lastCheckedAt)
220-
.where(sql`${t.deliveryStatus} in ('pending', 'delivering')`),
221-
// Tenant columns must lead so the index, not the `chats` join, narrows to this
222-
// user first. The trailing expression is what the wake queries filter and order on.
223-
index("watches_org_user_wake_idx")
224-
.on(t.organizationId, t.userId, sql`coalesce(${t.firedAt}, ${t.lastCheckedAt}) desc`)
225-
.where(sql`${t.deliveryStatus} = 'delivered' and ${t.status} in ('fired', 'expired')`),
226-
// Covers `listActiveWatchesForBatch`.
227-
index("watches_active_env_idx")
228-
.on(t.environmentId, t.expiresAt)
229-
.where(sql`${t.status} = 'active'`),
230-
]
231-
);
232-
233-
export type WatchBatchStatus = "running" | "stopped";
234-
235-
/**
236-
* `epoch` and `generation` are claimed together by {@link claimWatchBatchTick}, so a
237-
* duplicated schedule can't fork the chain. FK-free.
238-
*/
239-
export const watchBatches = dashboardAgentSchema.table(
240-
"watch_batches",
241-
{
242-
environmentId: text("environment_id").notNull(),
243-
// 1 | 5 | 15 | 60.
244-
cadenceMinutes: integer("cadence_minutes").notNull(),
245-
// Bumped by every arm. Runs carry it, and a claim requires it to match.
246-
epoch: integer("epoch").notNull().default(0),
247-
// Inside the current epoch. The claim is its only writer.
248-
generation: integer("generation").notNull().default(0),
249-
status: text("status").$type<WatchBatchStatus>().notNull().default("running"),
250-
armedAt: timestamp("armed_at", { withTimezone: true }).notNull().defaultNow(),
251-
// Heartbeat. NULL between an arm and the first run landing.
252-
lastTickAt: timestamp("last_tick_at", { withTimezone: true }),
253-
},
254-
(t) => [primaryKey({ columns: [t.environmentId, t.cadenceMinutes] })]
255-
);
256-
257142
export type Chat = typeof chats.$inferSelect;
258143
export type NewChat = typeof chats.$inferInsert;
259144
export type ChatSession = typeof chatSessions.$inferSelect;
@@ -262,6 +147,3 @@ export type ChatTurnEval = typeof chatTurnEvals.$inferSelect;
262147
export type NewChatTurnEval = typeof chatTurnEvals.$inferInsert;
263148
export type Investigation = typeof investigations.$inferSelect;
264149
export type NewInvestigation = typeof investigations.$inferInsert;
265-
export type Watch = typeof watches.$inferSelect;
266-
export type NewWatch = typeof watches.$inferInsert;
267-
export type WatchBatch = typeof watchBatches.$inferSelect;

‎internal-packages/dashboard-agent-db/src/watch-queries.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,16 +7,16 @@ import {
77
import type { DashboardAgentDb } from "./client.js";
88
import { generateWatchDeliveryClaimId, generateWatchId } from "./ids.js";
99
import { lockChatForWatches, type DashboardAgentDbOrTx } from "./internal.js";
10+
import { chats } from "./schema.js";
1011
import {
11-
chats,
1212
watchBatches,
1313
watches,
1414
type PersistedWatchSpec,
1515
type Watch,
1616
type WatchBatch,
1717
type WatchCancelReason,
1818
type WatchStatus,
19-
} from "./schema.js";
19+
} from "./watch-schema.js";
2020

2121
// The watch, wake and batch-chain half of the query layer. Same tenancy rule as
2222
// `queries.ts`: every read is scoped by organization and/or user.

‎internal-packages/dashboard-agent-db/src/watch-schema.ts‎

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
import { sql } from "drizzle-orm";
2+
import {
3+
boolean,
4+
index,
5+
integer,
6+
jsonb,
7+
primaryKey,
8+
text,
9+
timestamp,
10+
uniqueIndex,
11+
} from "drizzle-orm/pg-core";
12+
import type {
13+
WatchObservedOutcome,
14+
WatchResolution,
15+
WatchSpec,
16+
} from "@internal/dashboard-agent-contracts";
17+
import { dashboardAgentSchema } from "./schema-base.js";
18+
19+
// The watch tables. Re-exported by `schema.ts`, which is what drizzle-kit reads.
20+
21+
/** `active` is the only non-terminal status; the other three are immutable. */
22+
export type WatchStatus = "active" | "fired" | "expired" | "cancelled";
23+
/**
24+
* `not_required` while active and for every cancellation. `delivering` is the
25+
* in-flight claim, one deliverer at a time. See `claimWatchDelivery`.
26+
*/
27+
export type WatchDeliveryStatus = "not_required" | "pending" | "delivering" | "delivered";
28+
/** Cancellations are silent: no resolution, no wake. */
29+
export type WatchCancelReason = "user" | "access_revoked" | "chat_deleted" | "scheduling_failed";
30+
31+
/** `since` is server-set at creation, so `error_recurrence` can't match older errors. */
32+
export type PersistedWatchSpec = WatchSpec & { since?: string };
33+
34+
/**
35+
* The initiating identity is snapshotted at creation, so a membership change can only
36+
* cancel a watch, never widen its scope. Main-DB ids, FK-free.
37+
*/
38+
export const watches = dashboardAgentSchema.table(
39+
"watches",
40+
{
41+
id: text("id").primaryKey(), // = watchId (`watch_…`)
42+
chatId: text("chat_id").notNull(), // = chats.id
43+
identity: text("identity").notNull(),
44+
spec: jsonb("spec").$type<PersistedWatchSpec>().notNull(),
45+
status: text("status").$type<WatchStatus>().notNull().default("active"),
46+
deliveryStatus: text("delivery_status")
47+
.$type<WatchDeliveryStatus>()
48+
.notNull()
49+
.default("not_required"),
50+
cancelReason: text("cancel_reason").$type<WatchCancelReason>(),
51+
/**
52+
* The meaning; `status` above stays the two-value transport encoding so persisted
53+
* wake ids and dedup keys remain valid. NULL while active and on cancellation.
54+
*/
55+
resolution: text("resolution").$type<WatchResolution>(),
56+
/** Written in the same statement as `resolution` and `lastResult`. */
57+
observedOutcome: jsonb("observed_outcome").$type<WatchObservedOutcome>(),
58+
/** Consent given at creation. Never part of `identity`. */
59+
investigateOnAttention: boolean("investigate_on_attention").notNull().default(false),
60+
// Immutable initiating identity, snapshotted at creation.
61+
organizationId: text("organization_id").notNull(),
62+
projectId: text("project_id").notNull(),
63+
/** Nullable: rows created before this column don't carry it. */
64+
projectRef: text("project_ref"),
65+
environmentId: text("environment_id").notNull(),
66+
userId: text("user_id").notNull(),
67+
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
68+
expiresAt: timestamp("expires_at", { withTimezone: true }).notNull(),
69+
lastCheckedAt: timestamp("last_checked_at", { withTimezone: true }),
70+
firedAt: timestamp("fired_at", { withTimezone: true }),
71+
// A claim older than the delivery grace is abandoned and may be re-claimed.
72+
deliveryClaimedAt: timestamp("delivery_claimed_at", { withTimezone: true }),
73+
// Fencing token, written fresh on every claim and required by the release and
74+
// delivered marks, so a revived deliverer can't touch the claim that replaced it.
75+
deliveryClaimId: text("delivery_claim_id"),
76+
deliveredAt: timestamp("delivered_at", { withTimezone: true }),
77+
cancelledAt: timestamp("cancelled_at", { withTimezone: true }),
78+
// On fire or expire this is the notification's payload.
79+
lastResult: jsonb("last_result").$type<Record<string, unknown>>(),
80+
// Check idempotency keys are `watch:{id}:tick:{n}`.
81+
tickCount: integer("tick_count").notNull().default(0),
82+
},
83+
(t) => [
84+
index("watches_chat_idx").on(t.chatId),
85+
// UNIQUE is the dedup guarantee: a read-then-insert check can't be race-proof
86+
// under READ COMMITTED. Leading `chat_id` also serves the active-watches lookup.
87+
uniqueIndex("watches_chat_active_identity_key")
88+
.on(t.chatId, t.projectId, t.environmentId, t.identity)
89+
.where(sql`${t.status} = 'active'`),
90+
// Sweep: active watches due to be checked / past their expiry.
91+
index("watches_status_expires_idx").on(t.status, t.expiresAt),
92+
// Sweep: resolved watches whose wake is still owed.
93+
index("watches_pending_delivery_idx")
94+
.on(t.firedAt, t.lastCheckedAt)
95+
.where(sql`${t.deliveryStatus} in ('pending', 'delivering')`),
96+
// Tenant columns must lead so the index, not the `chats` join, narrows to this
97+
// user first. The trailing expression is what the wake queries filter and order on.
98+
index("watches_org_user_wake_idx")
99+
.on(t.organizationId, t.userId, sql`coalesce(${t.firedAt}, ${t.lastCheckedAt}) desc`)
100+
.where(sql`${t.deliveryStatus} = 'delivered' and ${t.status} in ('fired', 'expired')`),
101+
// Covers `listActiveWatchesForBatch`.
102+
index("watches_active_env_idx")
103+
.on(t.environmentId, t.expiresAt)
104+
.where(sql`${t.status} = 'active'`),
105+
]
106+
);
107+
108+
export type WatchBatchStatus = "running" | "stopped";
109+
110+
/**
111+
* `epoch` and `generation` are claimed together by {@link claimWatchBatchTick}, so a
112+
* duplicated schedule can't fork the chain. FK-free.
113+
*/
114+
export const watchBatches = dashboardAgentSchema.table(
115+
"watch_batches",
116+
{
117+
environmentId: text("environment_id").notNull(),
118+
// 1 | 5 | 15 | 60.
119+
cadenceMinutes: integer("cadence_minutes").notNull(),
120+
// Bumped by every arm. Runs carry it, and a claim requires it to match.
121+
epoch: integer("epoch").notNull().default(0),
122+
// Inside the current epoch. The claim is its only writer.
123+
generation: integer("generation").notNull().default(0),
124+
status: text("status").$type<WatchBatchStatus>().notNull().default("running"),
125+
armedAt: timestamp("armed_at", { withTimezone: true }).notNull().defaultNow(),
126+
// Heartbeat. NULL between an arm and the first run landing.
127+
lastTickAt: timestamp("last_tick_at", { withTimezone: true }),
128+
},
129+
(t) => [primaryKey({ columns: [t.environmentId, t.cadenceMinutes] })]
130+
);
131+
132+
export type Watch = typeof watches.$inferSelect;
133+
export type NewWatch = typeof watches.$inferInsert;
134+
export type WatchBatch = typeof watchBatches.$inferSelect;

0 commit comments

Comments
 (0)