Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | 27x 27x 305x 107x 107x 71x 11x 11x 10x 10x 904x 4x 5x 5x 5x 1011x | import { and, desc, eq, gte, sql } from "drizzle-orm";
import type { DrizzleD1Database } from "drizzle-orm/d1";
import * as schema from "../../db/schema";
import type {
RRM_ACTOR_TYPES,
RRM_CHANNELS,
RRM_EVENT_TYPES,
} from "../../db/schema/enums";
import type {
NewRrmProspectEvent,
RrmProspectEvent,
} from "../../db/schema/rrm";
import { generateId } from "../../lib/ids";
const EVENTS = schema.rrmProspectEvents;
export type RrmEventType = (typeof RRM_EVENT_TYPES)[number];
export type RrmActorType = (typeof RRM_ACTOR_TYPES)[number];
export type RrmChannel = (typeof RRM_CHANNELS)[number];
export type AppendEventInput = {
prospectId: string;
type: RrmEventType;
actorType: RrmActorType;
/** Null for events with no channel of their own, e.g. `imported`. */
channel?: RrmChannel | null;
payload?: Record<string, unknown> | null;
/** Operator user id, or the wamid/system name behind a non-human actor. */
actorId?: string | null;
/**
* Defaults to now. Pass it explicitly when the event has to carry the same
* instant as the row it describes — a stage change writes `stage_changed_at`
* and its `stage_changed` event in one batch, and two `new Date()` calls a
* millisecond apart would make the pair look inconsistent in the timeline.
*/
occurredAt?: Date;
};
/** Newest first; ties broken by insertion order. */
const DEFAULT_TIMELINE_LIMIT = 100;
/**
* `rrm_prospect_events` is APPEND-ONLY. There is deliberately no update and no
* delete on this class, and none may be added.
*
* Every funnel number, every ramp gate and the auto-halt are computed by
* counting these rows, never by reading a derived column on the prospect. A
* mutable event table would make every historical figure unreproducible: an
* edited `message_sent` silently changes last week's reply rate, and a deleted
* `opted_out` erases the evidence that we were asked to stop. Corrections are
* made by appending a further event, which is what an audit trail is for.
*
* The one erasure path — a DPDP deletion request — nulls the prospect row and
* appends `deleted`; it does not touch this table, which is why the table has
* no FK to prospects.
*/
export class RrmEventsDal {
constructor(private db: DrizzleD1Database<typeof schema>) {}
/** Appends one event and returns the stored row. */
async append(input: AppendEventInput): Promise<RrmProspectEvent> {
const rows = await this.db
.insert(EVENTS)
.values(this.toRow(input))
.returning();
return rows[0];
}
/**
* Appends many events atomically via `db.batch` — one statement per row.
*
* Not a single multi-row INSERT: eight columns puts a 13-row insert past
* D1's 100-bound-parameter ceiling, and an importer writing an `imported`
* event per prospect goes well past that.
*
* @returns the number of rows written.
*/
async appendMany(inputs: AppendEventInput[]): Promise<number> {
const statements = inputs.map((input) => this.buildEventStatement(input));
const [first, ...rest] = statements;
// `db.batch` demands a non-empty tuple, and an empty append is a no-op
// rather than a caller error — enrolling zero prospects is legitimate.
if (!first) return 0;
await this.db.batch([first, ...rest]);
return statements.length;
}
/**
* Builds the insert WITHOUT executing it, so another DAL can put the event
* inside its own `db.batch` alongside the row the event describes.
*
* This exists to make the spec's §0 rule 7 enforceable: "no stage is set by
* a UI click that isn't also an event row". The prospects DAL batches the
* `rrm_prospects` update and the `stage_changed` event together, so D1
* either applies both or neither. Executing the event separately would let
* a stage land with no event behind it, and the funnel would quietly under-
* count from then on with nothing to reconstruct it from.
*
* Drizzle query builders are lazy — nothing hits the database until the
* statement is awaited or handed to `batch`.
*/
buildEventStatement(input: AppendEventInput) {
return this.db.insert(EVENTS).values(this.toRow(input));
}
/** One prospect's history, newest first. */
async timelineFor(
prospectId: string,
options: { limit?: number } = {},
): Promise<RrmProspectEvent[]> {
return await this.db
.select()
.from(EVENTS)
.where(eq(EVENTS.prospectId, prospectId))
// `occurred_at` is second-precision, so a webhook that writes three
// events in one request gives them all the same timestamp. rowid is
// the insertion order, which is the real order those events happened
// in; without it the timeline shows them shuffled at random.
.orderBy(desc(EVENTS.occurredAt), sql`rowid desc`)
.limit(options.limit ?? DEFAULT_TIMELINE_LIMIT);
}
/**
* Counts events of one type, optionally since an instant.
*
* Feeds both the funnel and the auto-halt — the halt compares
* `countByType("send_blocked", todayStart)` against the day's sends, so
* `since` is a window boundary, not a convenience filter.
*/
async countByType(type: RrmEventType, since?: Date): Promise<number> {
const filter = since
? and(eq(EVENTS.type, type), gte(EVENTS.occurredAt, since))
: eq(EVENTS.type, type);
const rows = await this.db
.select({ count: sql<number>`count(*)` })
.from(EVENTS)
.where(filter);
return rows[0]?.count ?? 0;
}
/** Single place the id is minted and `occurredAt` defaulted. */
private toRow(input: AppendEventInput): NewRrmProspectEvent {
return {
id: generateId(),
prospectId: input.prospectId,
type: input.type,
channel: input.channel ?? null,
payload: input.payload ?? null,
actorType: input.actorType,
actorId: input.actorId ?? null,
occurredAt: input.occurredAt ?? new Date(),
};
}
}
/**
* The unexecuted insert `buildEventStatement` hands back. Assignable to
* `BatchItem<"sqlite">`, so a caller can drop it straight into `db.batch`.
*/
export type RrmEventStatement = ReturnType<RrmEventsDal["buildEventStatement"]>;
|