feat: initial Signage 0.1.0
This commit is contained in:
ekklesia
@@ -0,0 +1,371 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { ModuleDatabase } from "@ekklesia/module-sdk";
|
||||
import { and, asc, desc, eq, inArray, lt, sql } from "drizzle-orm";
|
||||
import {
|
||||
channelGroups,
|
||||
channels,
|
||||
playbackStates,
|
||||
selections,
|
||||
selectionVariants,
|
||||
} 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: ModuleDatabase): SignagePlaybackStore {
|
||||
async function variantsFor(
|
||||
selectionIds: readonly string[],
|
||||
): Promise<Map<string, StoredSelectionVariant[]>> {
|
||||
const grouped = new Map<string, StoredSelectionVariant[]>();
|
||||
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<StoredSelection[]> {
|
||||
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<StoredChannelGroupDetail | undefined> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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): Promise<void> {
|
||||
await database.delete(channelGroups).where(eq(channelGroups.id, groupId));
|
||||
},
|
||||
|
||||
async findChannelGroup(groupId: string): Promise<StoredChannelGroupDetail | undefined> {
|
||||
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<StoredSelection | undefined> {
|
||||
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<StoredSelection | undefined> {
|
||||
const rows = await database.select().from(selections).where(eq(selections.id, selectionId));
|
||||
return (await hydrateSelections(rows))[0];
|
||||
},
|
||||
|
||||
async insertSelection(selection: NewStoredSelection): Promise<void> {
|
||||
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<readonly StoredChannelGroupDetail[]> {
|
||||
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<readonly StoredSelection[]> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
const values: Record<string, unknown> = { 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<void> {
|
||||
await database.transaction(async (transaction) => {
|
||||
const values: Record<string, unknown> = {};
|
||||
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,
|
||||
})),
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user