Files
Signage/src/server/store/drizzle-timelines.ts
T
ekklesiaandClaude Opus 5.5 a6d553efbb
CI / lint (pull_request) Successful in 14s
CI / typecheck (pull_request) Successful in 16s
CI / test (pull_request) Successful in 21s
CI / build (pull_request) Successful in 20s
CI / integration (pull_request) Skipped
feat: move Signage to the Ekklesia 0.2 module contract (manifest v8, database ABI v1)
Signage moves from the legacy module contract to Ekklesia 0.2.0's, with no behaviour or schema
change:

- manifest version 8, `database: { abi: 1 }`, compatibility ^0.2.0 for platform and module SDK;
  the runtime declares the same ABI, as Core requires;
- every store reaches PostgreSQL through the platform's Module Database ABI v1 executor, with
  Signage's own Drizzle tables and the module-side adapter (`moduleDrizzle`) from
  @ekklesia/module-database. createSignageStore takes the executor, so no call site changes;
- the schema entrypoint exports only plain ABI v1 metadata (table names and foreign-key targets),
  derived from tables.ts with moduleDrizzleSchemaMetadata; no Drizzle object reaches Core;
- duplicate-image detection reads SQLSTATE 23505 from the stable Ekklesia database error, found
  with findEkklesiaDatabaseError through Drizzle's DrizzleQueryError wrapping. The old
  `error.code === "23505"` check never matched a wrapped error, so a duplicate-upload race
  surfaced as a 500 instead of the content-hash conflict.

tables.ts declarations, the migration, and its pinned checksum are unchanged.

Tests: an ABI v1 executor double that enforces Core's scope, parameter, and abort rules drives all
13 store transactions, their rollbacks, raw int8/timestamptz/jsonb decoding, and the 23505
regression, including generic and driver-shaped errors that must not count as duplicates.

CI, temporarily: the platform pin moves to v0.2.0 (1788d17) with a regenerated ci/pnpm-lock.yaml.
The integration job is disabled: from 0.2.0 the platform image is generic (ADR 0048) and installs
modules only as Module Packages, which Signage cannot be until its sharp dependency goes through
the native package flow. The job and ci/integration.sh stay, and the job returns in that change.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-08 20:36:11 -05:00

407 lines
14 KiB
TypeScript

import type {
ModuleDrizzleDatabase,
ModuleDrizzleTransaction,
} from "@ekklesia/module-database/drizzle";
import { asc, eq, or, sql } from "drizzle-orm";
import { playbackStates, providerSources, timelineSlots, timelines } from "../../schema/tables.js";
import {
type NewStoredProviderSource,
type NewStoredSlot,
type NewStoredTimeline,
SignageStoreConflict,
type SignageTimelineStore,
type SlotChanges,
type StoredProviderSource,
type StoredSlot,
type StoredTimeline,
type StoredTimelineDetail,
type StoredTimelineUsage,
type TimelineChanges,
} from "./types.js";
/**
* The timeline half of Signage's production store.
*
* Two invariants live here rather than in the services above, because they are only safe if they are
* transactional:
*
* - **Slot positions stay unique and contiguous from zero.** Every operation that can change the
* order rewrites the whole order inside one transaction, so a gap can never be observed — not even
* briefly, by a reconciliation running at the same moment.
* - **A timeline's revision increments exactly once per semantic change**, in the same transaction
* as the change, so a client that compares revisions can never see a changed timeline at an
* unchanged revision.
*/
type TimelineRow = typeof timelines.$inferSelect;
type SlotRow = typeof timelineSlots.$inferSelect;
type SourceRow = typeof providerSources.$inferSelect;
function toTimeline(row: TimelineRow): StoredTimeline {
return {
createdAt: row.createdAt,
description: row.description,
eligibleFrom: row.eligibleFrom,
eligibleUntil: row.eligibleUntil,
id: row.id,
name: row.name,
revision: row.revision,
updatedAt: row.updatedAt,
};
}
function toSource(row: SourceRow): StoredProviderSource {
return {
createdAt: row.createdAt,
id: row.id,
label: row.label,
providerId: row.providerId,
selectionScope: row.selectionScope,
sourceId: row.sourceId,
timelineId: row.timelineId,
};
}
function toSlot(row: SlotRow): StoredSlot {
return {
durationOverrideMs: row.durationOverrideMs,
eligibleFrom: row.eligibleFrom,
eligibleUntil: row.eligibleUntil,
enabled: row.enabled,
id: row.id,
itemId: row.itemId,
kind: row.kind === "pinned-item" ? "pinned-item" : "provider-source",
position: row.position,
presentation: {
backgroundColor: row.backgroundColor,
backgroundDarkened: row.backgroundDarkened,
backgroundMode:
row.backgroundMode === "blurred" ||
row.backgroundMode === "solid" ||
row.backgroundMode === "black"
? row.backgroundMode
: null,
blurPixels: row.blurPixels,
foregroundShadow: row.foregroundShadow,
},
providerId: row.providerId,
sourceId: row.sourceId,
timelineId: row.timelineId,
};
}
function slotValues(slot: NewStoredSlot, position: number) {
return {
backgroundColor: slot.presentation.backgroundColor,
backgroundDarkened: slot.presentation.backgroundDarkened,
backgroundMode: slot.presentation.backgroundMode,
blurPixels: slot.presentation.blurPixels,
durationOverrideMs: slot.durationOverrideMs,
eligibleFrom: slot.eligibleFrom,
eligibleUntil: slot.eligibleUntil,
enabled: slot.enabled,
foregroundShadow: slot.presentation.foregroundShadow,
id: slot.id,
itemId: slot.itemId,
kind: slot.kind,
position,
providerId: slot.providerId,
sourceId: slot.sourceId,
timelineId: slot.timelineId,
};
}
export function createTimelineStore(database: ModuleDrizzleDatabase): SignageTimelineStore {
/**
* Rewrites `position` for one timeline from an explicit order.
*
* Positions are written in two passes, through a temporary high range, because the unique
* `(timeline_id, position)` index is checked per statement: moving a slot from 3 to 1 while
* another still holds 1 would collide. Temporary positions sit beyond the final order while
* remaining inside the database's non-negative constraint.
*/
async function renumber(
transaction: ModuleDrizzleTransaction,
timelineId: string,
orderedSlotIds: readonly string[],
): Promise<void> {
const temporaryOffset = orderedSlotIds.length + 1;
for (const [index, slotId] of orderedSlotIds.entries()) {
await transaction
.update(timelineSlots)
.set({ position: temporaryOffset + index })
.where(eq(timelineSlots.id, slotId));
}
for (const [index, slotId] of orderedSlotIds.entries()) {
await transaction
.update(timelineSlots)
.set({ position: index, updatedAt: new Date() })
.where(eq(timelineSlots.id, slotId));
}
await bumpRevision(transaction, timelineId);
}
async function bumpRevision(
transaction: ModuleDrizzleTransaction,
timelineId: string,
): Promise<void> {
await transaction
.update(timelines)
.set({ revision: sql`${timelines.revision} + 1`, updatedAt: new Date() })
.where(eq(timelines.id, timelineId));
}
async function orderedSlotIdsFor(
transaction: ModuleDrizzleTransaction,
timelineId: string,
): Promise<string[]> {
const rows = await transaction
.select({ id: timelineSlots.id, position: timelineSlots.position })
.from(timelineSlots)
.where(eq(timelineSlots.timelineId, timelineId))
.orderBy(asc(timelineSlots.position));
return rows.map((row) => row.id);
}
async function timelineIdForSlot(
transaction: ModuleDrizzleTransaction,
slotId: string,
): Promise<string | undefined> {
const rows = await transaction
.select({ timelineId: timelineSlots.timelineId })
.from(timelineSlots)
.where(eq(timelineSlots.id, slotId));
return rows[0]?.timelineId;
}
return {
async addSlot(slot: NewStoredSlot, position?: number): Promise<void> {
await database.transaction(async (transaction) => {
const existing = await orderedSlotIdsFor(transaction, slot.timelineId);
const index =
position === undefined
? existing.length
: Math.max(0, Math.min(position, existing.length));
await transaction.insert(timelineSlots).values(slotValues(slot, existing.length + 1));
const next = [...existing.slice(0, index), slot.id, ...existing.slice(index)];
await renumber(transaction, slot.timelineId, next);
});
},
async addSource(source: NewStoredProviderSource): Promise<void> {
await database.insert(providerSources).values({
id: source.id,
label: source.label,
providerId: source.providerId,
selectionScope: source.selectionScope,
sourceId: source.sourceId,
timelineId: source.timelineId,
});
},
async createTimeline(input: {
readonly timeline: NewStoredTimeline;
readonly sources: readonly NewStoredProviderSource[];
readonly slots: readonly NewStoredSlot[];
}): Promise<void> {
await database.transaction(async (transaction) => {
await transaction.insert(timelines).values({
description: input.timeline.description,
eligibleFrom: input.timeline.eligibleFrom,
eligibleUntil: input.timeline.eligibleUntil,
id: input.timeline.id,
name: input.timeline.name,
});
if (input.sources.length > 0) {
await transaction.insert(providerSources).values(
input.sources.map((source) => ({
id: source.id,
label: source.label,
providerId: source.providerId,
selectionScope: source.selectionScope,
sourceId: source.sourceId,
timelineId: source.timelineId,
})),
);
}
if (input.slots.length > 0) {
await transaction
.insert(timelineSlots)
.values(input.slots.map((slot, index) => slotValues(slot, index)));
}
});
},
async deleteTimeline(timelineId: string): Promise<void> {
// Sources and slots cascade with the timeline; resolved selections do not, because history is
// a record of what a screen actually showed.
await database.delete(timelines).where(eq(timelines.id, timelineId));
},
async findTimeline(timelineId: string): Promise<StoredTimelineDetail | undefined> {
const timelineRows = await database
.select()
.from(timelines)
.where(eq(timelines.id, timelineId));
const row = timelineRows[0];
if (!row) {
return undefined;
}
const [sourceRows, slotRows] = await Promise.all([
database
.select()
.from(providerSources)
.where(eq(providerSources.timelineId, timelineId))
.orderBy(asc(providerSources.createdAt)),
database
.select()
.from(timelineSlots)
.where(eq(timelineSlots.timelineId, timelineId))
.orderBy(asc(timelineSlots.position)),
]);
return {
slots: slotRows.map(toSlot),
sources: sourceRows.map(toSource),
timeline: toTimeline(row),
};
},
async listTimelines(): Promise<readonly StoredTimeline[]> {
const rows = await database.select().from(timelines).orderBy(asc(timelines.name));
return rows.map(toTimeline);
},
async removeSlot(slotId: string): Promise<void> {
await database.transaction(async (transaction) => {
const timelineId = await timelineIdForSlot(transaction, slotId);
if (timelineId === undefined) {
return;
}
await transaction.delete(timelineSlots).where(eq(timelineSlots.id, slotId));
await renumber(transaction, timelineId, await orderedSlotIdsFor(transaction, timelineId));
});
},
async removeSource(sourceId: string): Promise<void> {
await database.transaction(async (transaction) => {
const rows = await transaction
.select({ timelineId: providerSources.timelineId })
.from(providerSources)
.where(eq(providerSources.id, sourceId));
const timelineId = rows[0]?.timelineId;
if (timelineId === undefined) {
return;
}
// Slots referencing the source cascade with it; the remainder is renumbered so the pattern
// an operator reads never has a hole in it.
await transaction.delete(providerSources).where(eq(providerSources.id, sourceId));
await renumber(transaction, timelineId, await orderedSlotIdsFor(transaction, timelineId));
});
},
async reorderSlots(timelineId: string, orderedSlotIds: readonly string[]): Promise<void> {
await database.transaction(async (transaction) => {
const current = await orderedSlotIdsFor(transaction, timelineId);
const given = new Set(orderedSlotIds);
// Exhaustive and exact: anything else could lose a slot, duplicate one, or leave a gap.
if (given.size !== orderedSlotIds.length || given.size !== current.length) {
throw new SignageStoreConflict("timeline.slotOrder");
}
for (const slotId of current) {
if (!given.has(slotId)) {
throw new SignageStoreConflict("timeline.slotOrder");
}
}
await renumber(transaction, timelineId, orderedSlotIds);
});
},
async timelineUsage(timelineId: string): Promise<StoredTimelineUsage> {
const rows = await database
.select({
activeTimelineId: playbackStates.activeTimelineId,
pendingTimelineId: playbackStates.pendingTimelineId,
scheduledTimelineId: playbackStates.scheduledTimelineId,
})
.from(playbackStates)
.where(
or(
eq(playbackStates.activeTimelineId, timelineId),
eq(playbackStates.pendingTimelineId, timelineId),
eq(playbackStates.scheduledTimelineId, timelineId),
),
);
return {
activeGroups: rows.filter((row) => row.activeTimelineId === timelineId).length,
pendingGroups: rows.filter((row) => row.pendingTimelineId === timelineId).length,
scheduledGroups: rows.filter((row) => row.scheduledTimelineId === timelineId).length,
};
},
async updateSlot(slotId: string, changes: SlotChanges): Promise<void> {
await database.transaction(async (transaction) => {
const timelineId = await timelineIdForSlot(transaction, slotId);
if (timelineId === undefined) {
return;
}
const values: Record<string, unknown> = { updatedAt: new Date() };
if (changes.enabled !== undefined) {
values.enabled = changes.enabled;
}
if (changes.durationOverrideMs !== undefined) {
values.durationOverrideMs = changes.durationOverrideMs;
}
if (changes.eligibleFrom !== undefined) {
values.eligibleFrom = changes.eligibleFrom;
}
if (changes.eligibleUntil !== undefined) {
values.eligibleUntil = changes.eligibleUntil;
}
if (changes.itemId !== undefined) {
values.itemId = changes.itemId;
}
const presentation = changes.presentation;
if (presentation) {
if (presentation.backgroundMode !== undefined) {
values.backgroundMode = presentation.backgroundMode;
}
if (presentation.backgroundColor !== undefined) {
values.backgroundColor = presentation.backgroundColor;
}
if (presentation.backgroundDarkened !== undefined) {
values.backgroundDarkened = presentation.backgroundDarkened;
}
if (presentation.blurPixels !== undefined) {
values.blurPixels = presentation.blurPixels;
}
if (presentation.foregroundShadow !== undefined) {
values.foregroundShadow = presentation.foregroundShadow;
}
}
await transaction.update(timelineSlots).set(values).where(eq(timelineSlots.id, slotId));
await bumpRevision(transaction, timelineId);
});
},
async updateTimeline(timelineId: string, changes: TimelineChanges): Promise<void> {
const values: Record<string, unknown> = {
revision: sql`${timelines.revision} + 1`,
updatedAt: new Date(),
};
if (changes.name !== undefined) {
values.name = changes.name;
}
if (changes.description !== undefined) {
values.description = changes.description;
}
if (changes.eligibleFrom !== undefined) {
values.eligibleFrom = changes.eligibleFrom;
}
if (changes.eligibleUntil !== undefined) {
values.eligibleUntil = changes.eligibleUntil;
}
await database.update(timelines).set(values).where(eq(timelines.id, timelineId));
},
};
}