import { randomUUID } from "node:crypto"; import type { ModuleDrizzleDatabase } from "@ekklesia/module-database/drizzle"; import { and, asc, desc, eq, inArray, lt, sql } from "drizzle-orm"; import { channelGroups, channels, playbackStates, selections, selectionVariants, viewers, } from "../../schema/tables.js"; import { type NewStoredSelection, type PlaybackChanges, type SelectionChanges, type SignagePlaybackStore, SignageStoreConflict, type StoredChannel, type StoredChannelGroupDetail, type StoredPlaybackState, type StoredSelection, type StoredSelectionVariant, } from "./types.js"; /** * The playback half of Signage's production store. * * Two things here are transactional because they cannot be safe otherwise: creating a group together * with its playback row and its default channel, and taking the next selection sequence. A group * without a default channel has nowhere for a viewer to land, and two selections that took the same * sequence would make history ambiguous exactly where `previous` reads it. */ type SelectionRow = typeof selections.$inferSelect; function toPlayback(row: typeof playbackStates.$inferSelect): StoredPlaybackState { return { activeTimelineId: row.activeTimelineId, blackout: row.blackout, currentSelectionId: row.currentSelectionId, currentSlotId: row.currentSlotId, groupId: row.groupId, nextSequence: row.nextSequence, paused: row.paused, pausedRemainingMs: row.pausedRemainingMs, pendingRequestedAt: row.pendingRequestedAt, pendingTimelineId: row.pendingTimelineId, revision: row.revision, scheduledActivationAt: row.scheduledActivationAt, scheduledTimelineId: row.scheduledTimelineId, slotDeadlineAt: row.slotDeadlineAt, slotStartedAt: row.slotStartedAt, staticAssetId: row.staticAssetId, updatedAt: row.updatedAt, }; } function toChannel(row: typeof channels.$inferSelect): StoredChannel { return { groupId: row.groupId, id: row.id, isDefault: row.isDefault, key: row.key, name: row.name, }; } function toSelection( row: SelectionRow, variants: readonly StoredSelectionVariant[], ): StoredSelection { return { availableUntil: row.availableUntil, deadlineAt: row.deadlineAt, durationMs: row.durationMs, groupId: row.groupId, id: row.id, itemId: row.itemId, itemRevision: row.itemRevision, presentation: row.presentation, providerId: row.providerId, rendererContributionId: row.rendererContributionId, rendererContributionVersion: row.rendererContributionVersion, resolvedAt: row.resolvedAt, revision: row.revision, selectionScope: row.selectionScope, sequence: row.sequence, slotId: row.slotId, stale: row.stale, startedAt: row.startedAt, timelineId: row.timelineId, variants, withdrawn: row.withdrawn, }; } export function createPlaybackStore(database: ModuleDrizzleDatabase): SignagePlaybackStore { async function variantsFor( selectionIds: readonly string[], ): Promise> { const grouped = new Map(); if (selectionIds.length === 0) { return grouped; } const rows = await database .select() .from(selectionVariants) .where(inArray(selectionVariants.selectionId, [...selectionIds])); for (const row of rows) { const list = grouped.get(row.selectionId) ?? []; list.push({ audienceKey: row.audienceKey, isDefault: row.isDefault, payload: row.payload }); grouped.set(row.selectionId, list); } return grouped; } async function hydrateSelections(rows: readonly SelectionRow[]): Promise { const variants = await variantsFor(rows.map((row) => row.id)); return rows.map((row) => toSelection(row, variants.get(row.id) ?? [])); } async function groupDetail( groupRow: typeof channelGroups.$inferSelect, ): Promise { const [channelRows, playbackRows] = await Promise.all([ database .select() .from(channels) .where(eq(channels.groupId, groupRow.id)) .orderBy(desc(channels.isDefault), asc(channels.key)), database.select().from(playbackStates).where(eq(playbackStates.groupId, groupRow.id)), ]); const playback = playbackRows[0]; if (!playback) { return undefined; } return { channels: channelRows.map(toChannel), group: { id: groupRow.id, name: groupRow.name, revision: groupRow.revision }, playback: toPlayback(playback), }; } return { async addChannel(channel: StoredChannel): Promise { await database.transaction(async (transaction) => { await transaction.insert(channels).values({ groupId: channel.groupId, id: channel.id, isDefault: channel.isDefault, key: channel.key, name: channel.name, }); await transaction .update(channelGroups) .set({ revision: sql`${channelGroups.revision} + 1`, updatedAt: new Date() }) .where(eq(channelGroups.id, channel.groupId)); }); }, async createChannelGroup(input: { readonly groupId: string; readonly name: string; readonly channelId: string; readonly channelName: string; }): Promise { await database.transaction(async (transaction) => { await transaction.insert(channelGroups).values({ id: input.groupId, name: input.name }); await transaction.insert(channels).values({ groupId: input.groupId, id: input.channelId, isDefault: true, key: "default", name: input.channelName, }); await transaction.insert(playbackStates).values({ groupId: input.groupId }); }); }, async deleteChannelGroup(groupId: string, fallbackChannelId: string): Promise { await database.transaction(async (transaction) => { const removedChannels = await transaction .select({ id: channels.id }) .from(channels) .where(eq(channels.groupId, groupId)); const removedChannelIds = removedChannels.map((channel) => channel.id); if (removedChannelIds.length > 0) { await transaction .update(viewers) .set({ channelId: fallbackChannelId, updatedAt: new Date() }) .where(inArray(viewers.channelId, removedChannelIds)); } await transaction.delete(channelGroups).where(eq(channelGroups.id, groupId)); }); }, async findChannelGroup(groupId: string): Promise { const rows = await database.select().from(channelGroups).where(eq(channelGroups.id, groupId)); const row = rows[0]; return row ? groupDetail(row) : undefined; }, async findLatestSelectionForScope( groupId: string, selectionScope: string, ): Promise { const rows = await database .select() .from(selections) .where(and(eq(selections.groupId, groupId), eq(selections.selectionScope, selectionScope))) .orderBy(desc(selections.sequence)) .limit(1); return (await hydrateSelections(rows))[0]; }, async findSelection(selectionId: string): Promise { const rows = await database.select().from(selections).where(eq(selections.id, selectionId)); return (await hydrateSelections(rows))[0]; }, async insertSelection(selection: NewStoredSelection): Promise { await database.transaction(async (transaction) => { await transaction.insert(selections).values({ availableUntil: selection.availableUntil, deadlineAt: selection.deadlineAt, durationMs: selection.durationMs, groupId: selection.groupId, id: selection.id, itemId: selection.itemId, itemRevision: selection.itemRevision, presentation: selection.presentation, providerId: selection.providerId, rendererContributionId: selection.rendererContributionId, rendererContributionVersion: selection.rendererContributionVersion, selectionScope: selection.selectionScope, sequence: selection.sequence, slotId: selection.slotId, stale: selection.stale, startedAt: selection.startedAt, timelineId: selection.timelineId, withdrawn: selection.withdrawn, }); if (selection.variants.length > 0) { await transaction.insert(selectionVariants).values( selection.variants.map((variant) => ({ audienceKey: variant.audienceKey, id: randomUUID(), isDefault: variant.isDefault, payload: variant.payload, selectionId: selection.id, })), ); } }); }, async listChannelGroups(): Promise { const rows = await database.select().from(channelGroups).orderBy(asc(channelGroups.name)); const details = await Promise.all(rows.map((row) => groupDetail(row))); return details.filter((detail): detail is StoredChannelGroupDetail => detail !== undefined); }, async listSelections(groupId: string, limit: number): Promise { const rows = await database .select() .from(selections) .where(eq(selections.groupId, groupId)) .orderBy(desc(selections.sequence)) .limit(limit); return hydrateSelections(rows); }, async pruneSelections(groupId: string, limit: number): Promise { const keep = await database .select({ sequence: selections.sequence }) .from(selections) .where(eq(selections.groupId, groupId)) .orderBy(desc(selections.sequence)) .limit(limit); const oldest = keep.at(-1)?.sequence; if (oldest === undefined) { return; } await database .delete(selections) .where(and(eq(selections.groupId, groupId), lt(selections.sequence, oldest))); }, async removeChannel(channelId: string): Promise { await database.transaction(async (transaction) => { const rows = await transaction.select().from(channels).where(eq(channels.id, channelId)); const channel = rows[0]; if (!channel) { return; } if (channel.isDefault) { // Something must remain for a viewer to land on. throw new SignageStoreConflict("channel.default"); } await transaction.delete(channels).where(eq(channels.id, channelId)); await transaction .update(channelGroups) .set({ revision: sql`${channelGroups.revision} + 1`, updatedAt: new Date() }) .where(eq(channelGroups.id, channel.groupId)); }); }, async takeSelectionSequence(groupId: string): Promise { const rows = await database .update(playbackStates) .set({ nextSequence: sql`${playbackStates.nextSequence} + 1` }) .where(eq(playbackStates.groupId, groupId)) .returning({ nextSequence: playbackStates.nextSequence }); const next = rows[0]?.nextSequence; if (next === undefined) { throw new SignageStoreConflict("playback.sequence"); } // `returning` gives the incremented value, so the sequence this call owns is one below it. return next - 1; }, async updatePlayback( groupId: string, changes: PlaybackChanges, options?: { readonly bumpRevision?: boolean }, ): Promise { const values: Record = { updatedAt: new Date() }; for (const [key, value] of Object.entries(changes)) { if (value !== undefined) { values[key] = value; } else if (Object.hasOwn(changes, key)) { values[key] = null; } } if (options?.bumpRevision !== false) { values.revision = sql`${playbackStates.revision} + 1`; } await database.update(playbackStates).set(values).where(eq(playbackStates.groupId, groupId)); }, async updateSelection(selectionId: string, changes: SelectionChanges): Promise { await database.transaction(async (transaction) => { const values: Record = {}; for (const key of [ "availableUntil", "deadlineAt", "durationMs", "itemRevision", "presentation", "stale", "startedAt", "withdrawn", ] as const) { if (changes[key] !== undefined) { values[key] = changes[key]; } } if (changes.bumpRevision === true) { values.revision = sql`${selections.revision} + 1`; } if (Object.keys(values).length > 0) { await transaction.update(selections).set(values).where(eq(selections.id, selectionId)); } if (changes.variants) { // Replaced wholesale: a partial update would leave one audience rendering a payload from // a revision that no longer exists. await transaction .delete(selectionVariants) .where(eq(selectionVariants.selectionId, selectionId)); if (changes.variants.length > 0) { await transaction.insert(selectionVariants).values( changes.variants.map((variant) => ({ audienceKey: variant.audienceKey, id: randomUUID(), isDefault: variant.isDefault, payload: variant.payload, selectionId, })), ); } } }); }, }; }