Signage/src/server/store/drizzle-playback.ts

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,
  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: 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, fallbackChannelId: string): Promise<void> {
      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<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,
              })),
            );
          }
        }
      });
    },
  };
}