WIP: Phase 3B playback engine handoff

This commit is contained in:
Media
2026-08-16 21:50:19 -05:00
parent e2ecdc97e2
commit 68273d457f
16 changed files with 3506 additions and 244 deletions
+55 -2
View File
@@ -110,17 +110,70 @@ server-evaluated eligibility, immutable display/thumbnail URLs, and the
settings revision and evaluation time used. Playlist revisions increment once
per semantic content change. Exact no-ops and reads do not increment them.
Before a playback-relevant playlist, settings, or asset mutation, the playback
service catches up using the old metadata. The domain transaction then commits
and playback synchronously reconciles the new values at the mutation instant.
This prevents a late timer or restart boundary from being rewritten by a later
edit. If the post-commit reconciliation unexpectedly fails, the API returns an
error even though the preceding domain commit may already be durable; clients
must refetch current state before deciding whether to retry.
Activation is explicit. A request is held in the existing pending-activation
state while a playable normal slide is mid-duration; otherwise the target
playlist is prepared immediately at its first currently playable entry. The
output-off flag remains only a visual gate, and static hold is preserved while
the hidden normal playlist changes. Phase 3A does not include the advancing
playback timer that will consume pending activation at a slide boundary.
the hidden normal playlist changes. The authoritative playback service consumes
a pending activation at the current slide boundary and never merges the old and
new playlists.
Socket.IO publishes `playlists:changed` as a refetch invalidation after
committed playlist changes and after settings or referenced-asset edits that
affect resolved playlist data. Socket.IO is not a mutation API.
## Authoritative playback
One serialized service owns persisted global playback, timer reconciliation,
and every playback control path. Its REST surface is:
- `GET /api/playback`
- `POST /api/playback/pause`
- `POST /api/playback/resume`
- `POST /api/playback/next`
- `POST /api/playback/previous`
- `POST /api/playback/static`
- `PUT /api/playback/mode`
- `POST /api/playback/reload`
Visual output precedence is mode off (black), a valid static asset, the current
normal entry, then black when no playable target exists. Mode off is only a
visual gate: normal playback continues underneath unless it is independently
paused or held by static mode. Pause and static both preserve the normal cursor
and remaining duration. Resume clears both static and paused state together and
continues the normal entry with its preserved remaining time. Next and previous
wrap across currently playable entries; they change the held entry without
unpausing when paused and are no-ops while static.
Normal slides use persisted absolute start times and deadlines. Chained
timeouts reconcile from the prior authoritative deadline, including late
callbacks and restart catch-up, so delays do not accumulate as timing drift.
The service also arms exact schedule and asset-expiration boundaries. It skips
disabled, out-of-schedule, expired, and non-ready entries, and safely returns to
an eligible entry when one becomes available. Pending playlist activation is
applied at the current normal slide boundary; an empty target playlist still
becomes active with black normal output.
Socket.IO sends `playback:state` immediately on connection and after meaningful
committed playback changes. The authoritative snapshot includes server time and
revisions, mode and transport state, active playlist and timeline data, current
and static targets, crossfade settings, and a display-cache manifest with
effective duration, presentation, scheduling, expiration, and immutable media
metadata. Enabled future-scheduled entries remain in the manifest; disabled
entries and private S3 details do not. Reads and connections do not increment
the playback revision.
Phase 3B exposes this state and behavior through REST and Socket.IO only. MQTT
command wiring and the browser display renderer remain later phases.
## Storage and health
The API uses a small path-style S3 adapter configured by `S3_ENDPOINT`,
+44 -5
View File
@@ -6,6 +6,8 @@ const {
const { createAssetRouter } = require("./assets/routes");
const { createMediaRouter } = require("./assets/media-routes");
const { createAssetService } = require("./assets/service");
const { createPlaybackService } = require("./playback");
const { createPlaybackRouter } = require("./playback/routes");
const {
createPlaylistRouter,
createPlaylistService,
@@ -13,10 +15,10 @@ const {
const { createSettingsRouter } = require("./settings/routes");
const { createSettingsService } = require("./settings/service");
function safeNotifier(notifier, logger) {
function safeNotifier(notifier, logger, failureMessage) {
const warn = (error) => {
try {
logger?.warn?.("Playlist invalidation notification failed", error);
logger?.warn?.(failureMessage, error);
} catch {
// A best-effort notification or warning must not undo a database commit.
}
@@ -40,10 +42,35 @@ function createApp({
migrations,
now = Date.now,
onPlaylistsChanged = () => {},
playbackService: suppliedPlaybackService,
storage,
}) {
const app = express();
const notifyPlaylistsChanged = safeNotifier(onPlaylistsChanged, logger);
const notifyPlaylistsChanged = safeNotifier(
onPlaylistsChanged,
logger,
"Playlist invalidation notification failed",
);
const playbackService =
suppliedPlaybackService ||
createPlaybackService({
database,
now,
onActivationChanged: notifyPlaylistsChanged,
});
if (!suppliedPlaybackService) {
playbackService.initialize();
}
const reconcilePlaybackDomain = (change) =>
playbackService.domainChanged(change);
const reconcileBeforeDomainChange = () =>
playbackService.reconcile("before_domain_change");
const notifyPlaylistMutation = (change) => {
reconcilePlaybackDomain(change);
const publicChange = { ...change };
delete publicChange.invalidBoundaryChanged;
notifyPlaylistsChanged(publicChange);
};
const settingsService = createSettingsService(database);
const assetService = createAssetService({
database,
@@ -53,8 +80,10 @@ function createApp({
const playlistService = createPlaylistService({
database,
now,
onChanged: notifyPlaylistsChanged,
onBeforeChange: reconcileBeforeDomainChange,
onChanged: notifyPlaylistMutation,
});
app.locals.playbackService = playbackService;
app.disable("x-powered-by");
app.set("trust proxy", config.trustProxy);
@@ -85,18 +114,28 @@ function createApp({
app.use(
"/api/settings",
createSettingsRouter(settingsService, {
onBeforePlaybackDomainChange: reconcileBeforeDomainChange,
onPlaylistsChanged: notifyPlaylistsChanged,
onPlaybackDomainChanged: reconcilePlaybackDomain,
}),
);
app.use(
"/api/assets",
createAssetRouter({
assetService,
onBeforePlaybackDomainChange: reconcileBeforeDomainChange,
onPlaylistsChanged: notifyPlaylistsChanged,
onPlaybackDomainChanged: reconcilePlaybackDomain,
processingConfig: config.processing,
}),
);
app.use("/api/playlists", createPlaylistRouter(playlistService));
app.use(
"/api/playlists",
createPlaylistRouter(playlistService, {
playbackService,
}),
);
app.use("/api/playback", createPlaybackRouter(playbackService));
app.use("/media", createMediaRouter(assetService));
app.use((request, response) => {
+18 -3
View File
@@ -8,7 +8,9 @@ const { sendOriginal } = require("./streaming");
function createAssetRouter({
assetService,
onBeforePlaybackDomainChange = () => {},
onPlaylistsChanged = () => {},
onPlaybackDomainChanged = () => {},
processingConfig,
}) {
const router = express.Router();
@@ -92,13 +94,26 @@ function createAssetRouter({
.playlists.map((reference) => reference.playlistId),
),
];
onBeforePlaybackDomainChange({
reason: "before_asset_update",
assetId: previous.id,
});
const updated = assetService.update(request.params.assetId, request.body);
if (updated.updatedAt !== previous.updatedAt && playlistIds.length > 0) {
onPlaylistsChanged({
if (updated.updatedAt !== previous.updatedAt) {
const change = {
reason: "asset_updated",
playlistIds,
assetId: updated.id,
});
...(updated.expirationAt !== previous.expirationAt
? { invalidBoundaryChanged: true }
: {}),
};
onPlaybackDomainChanged(change);
if (playlistIds.length > 0) {
const publicChange = { ...change };
delete publicChange.invalidBoundaryChanged;
onPlaylistsChanged(publicChange);
}
}
response.json(updated);
} catch (error) {
+29
View File
@@ -0,0 +1,29 @@
class PlaybackError extends Error {
constructor(code, message, { statusCode = 400, details } = {}) {
super(message);
this.name = "PlaybackError";
this.code = code;
this.statusCode = statusCode;
if (details !== undefined) this.details = details;
}
}
function playlistNotFound(playlistId) {
return new PlaybackError("PLAYLIST_NOT_FOUND", "Playlist not found", {
statusCode: 404,
details: { playlistId },
});
}
function assetNotFound(assetId) {
return new PlaybackError("ASSET_NOT_FOUND", "Library asset not found", {
statusCode: 404,
details: { assetId },
});
}
module.exports = {
PlaybackError,
assetNotFound,
playlistNotFound,
};
+7
View File
@@ -0,0 +1,7 @@
const { PlaybackError } = require("./errors");
const { createPlaybackService } = require("./service");
module.exports = {
PlaybackError,
createPlaybackService,
};
+75
View File
@@ -0,0 +1,75 @@
const express = require("express");
function createPlaybackRouter(playbackService) {
const router = express.Router();
router.get("/", (request, response, next) => {
try {
response.json(playbackService.getSnapshot());
} catch (error) {
next(error);
}
});
router.post("/pause", (request, response, next) => {
try {
response.json(playbackService.pause());
} catch (error) {
next(error);
}
});
router.post("/resume", (request, response, next) => {
try {
response.json(playbackService.resume());
} catch (error) {
next(error);
}
});
router.post("/next", (request, response, next) => {
try {
response.json(playbackService.next());
} catch (error) {
next(error);
}
});
router.post("/previous", (request, response, next) => {
try {
response.json(playbackService.previous());
} catch (error) {
next(error);
}
});
router.post("/static", (request, response, next) => {
try {
response.json(playbackService.enterStatic(request.body?.assetId));
} catch (error) {
next(error);
}
});
router.put("/mode", (request, response, next) => {
try {
response.json(playbackService.setMode(request.body?.mode));
} catch (error) {
next(error);
}
});
router.post("/reload", (request, response, next) => {
try {
response.json(playbackService.reload());
} catch (error) {
next(error);
}
});
return router;
}
module.exports = {
createPlaybackRouter,
};
File diff suppressed because it is too large Load Diff
+9 -2
View File
@@ -24,7 +24,10 @@ function orderedEntryIds(body) {
return body.entryIds;
}
function createPlaylistRouter(playlistService) {
function createPlaylistRouter(
playlistService,
{ playbackService } = {},
) {
const router = express.Router();
router.get("/", (request, response, next) => {
@@ -121,7 +124,11 @@ function createPlaylistRouter(playlistService) {
router.post("/:playlistId/activate", (request, response, next) => {
try {
response.json(playlistService.activate(request.params.playlistId));
if (!playbackService) {
throw new Error("The playback service is required for playlist activation");
}
const result = playbackService.activatePlaylist(request.params.playlistId);
response.json(result);
} catch (error) {
next(error);
}
+26 -197
View File
@@ -165,24 +165,6 @@ function mapPlaylist(row) {
};
}
function mapPlaybackState(row) {
return {
outputId: row.output_id,
mode: row.mode,
activePlaylistId: row.active_playlist_id,
pendingPlaylistId: row.pending_playlist_id,
pendingPlaylistRequestedAt: row.pending_playlist_requested_at,
currentEntryId: row.current_entry_id,
rotationPaused: Boolean(row.rotation_paused),
pausedRemainingMs: row.paused_remaining_ms,
staticAssetId: row.static_asset_id,
slideStartedAt: row.slide_started_at,
nextSlideAt: row.next_slide_at,
revision: row.revision,
updatedAt: row.updated_at,
};
}
function mediaUrl(assetId, kind, sha256) {
return sha256
? `/media/assets/${assetId}/${kind}/${sha256}.webp`
@@ -238,6 +220,7 @@ function createPlaylistService({
database,
now = Date.now,
createId = crypto.randomUUID,
onBeforeChange = () => {},
onChanged = () => {},
}) {
const playlistSelect = `
@@ -371,52 +354,6 @@ function createPlaylistService({
SET position = ?, updated_at = ?
WHERE playlist_id = ? AND id = ?
`);
const clearDeletedCurrentEntry = database.prepare(`
UPDATE playback_state
SET current_entry_id = NULL,
slide_started_at = NULL,
next_slide_at = NULL,
paused_remaining_ms = NULL,
revision = revision + 1,
updated_at = MAX(updated_at + 1, ?)
WHERE current_entry_id = ?
`);
const setPendingActivation = database.prepare(`
UPDATE playback_state
SET pending_playlist_id = ?,
pending_playlist_requested_at = ?,
revision = revision + 1,
updated_at = ?
WHERE output_id = 'global'
`);
const clearPendingActivation = database.prepare(`
UPDATE playback_state
SET pending_playlist_id = NULL,
pending_playlist_requested_at = NULL,
revision = revision + 1,
updated_at = ?
WHERE output_id = 'global'
`);
const activateImmediately = database.prepare(`
UPDATE playback_state
SET active_playlist_id = @playlistId,
pending_playlist_id = NULL,
pending_playlist_requested_at = NULL,
current_entry_id = @currentEntryId,
paused_remaining_ms = @pausedRemainingMs,
slide_started_at = @slideStartedAt,
next_slide_at = @nextSlideAt,
revision = revision + 1,
updated_at = @updatedAt
WHERE output_id = 'global'
`);
const findCurrentEntry = database.prepare(`
SELECT pe.*, a.status, a.expiration_at
FROM playlist_entries pe
JOIN assets a ON a.id = pe.asset_id
WHERE pe.id = ? AND pe.playlist_id = ?
`);
function getPlaylistRow(playlistId) {
const row = findPlaylist.get(playlistId);
if (!row) throw playlistNotFound(playlistId);
@@ -435,11 +372,10 @@ function createPlaylistService({
}
function notify(change) {
try {
onChanged(change);
} catch {
// The database commit is authoritative; invalidation must not undo it.
}
// The transaction has committed before this synchronous domain hook runs.
// A reconciliation failure is surfaced to the caller so playback cannot
// silently remain inconsistent with a successful playlist mutation.
onChanged(change);
}
function duplicateName(name, row = findPlaylistByName.get(name)) {
@@ -628,6 +564,7 @@ function createPlaylistService({
});
function update(playlistId, input) {
onBeforeChange({ reason: "before_playlist_update", playlistId });
const result = updateTransaction(playlistId, input);
if (result.changed) {
notify({
@@ -662,6 +599,7 @@ function createPlaylistService({
});
function remove(playlistId) {
onBeforeChange({ reason: "before_playlist_delete", playlistId });
const playlist = removeTransaction(playlistId);
notify({ reason: "deleted", playlistId, revision: playlist.revision });
return { id: playlistId, deleted: true };
@@ -743,6 +681,7 @@ function createPlaylistService({
});
function addEntry(playlistId, input) {
onBeforeChange({ reason: "before_entry_add", playlistId });
const result = addEntryTransaction(playlistId, input);
notify({
reason: "entry_added",
@@ -760,6 +699,7 @@ function createPlaylistService({
if (entryValuesEqual(current, fields)) {
return {
changed: false,
invalidBoundaryChanged: false,
playlist,
detail: resolvedDetail(playlistId, now()),
};
@@ -774,12 +714,19 @@ function createPlaylistService({
bumpPlaylistRevision.run(timestamp, playlistId);
return {
changed: true,
invalidBoundaryChanged:
current.schedule_end_at !== fields.scheduleEndAt,
playlist: getPlaylistRow(playlistId),
detail: resolvedDetail(playlistId, now()),
};
});
function updateEntryById(playlistId, entryId, input) {
onBeforeChange({
reason: "before_entry_update",
playlistId,
entryId,
});
const result = patchEntryTransaction(playlistId, entryId, input);
if (result.changed) {
notify({
@@ -787,6 +734,9 @@ function createPlaylistService({
playlistId,
revision: result.playlist.revision,
entryId,
...(result.invalidBoundaryChanged
? { invalidBoundaryChanged: true }
: {}),
});
}
return result.detail.entries.find((entry) => entry.id === entryId);
@@ -796,7 +746,6 @@ function createPlaylistService({
const playlist = getPlaylistRow(playlistId);
getEntryRow(playlistId, entryId);
const timestamp = nextTimestamp(playlist);
const playbackChanged = clearDeletedCurrentEntry.run(timestamp, entryId).changes > 0;
const result = deleteEntry.run(playlistId, entryId);
if (result.changes !== 1) throw playlistEntryNotFound(playlistId, entryId);
const remaining = listEntryIds.all(playlistId);
@@ -807,18 +756,21 @@ function createPlaylistService({
});
}
bumpPlaylistRevision.run(timestamp, playlistId);
return { playlist: getPlaylistRow(playlistId), playbackChanged };
return { playlist: getPlaylistRow(playlistId) };
});
function removeEntryById(playlistId, entryId) {
onBeforeChange({
reason: "before_entry_delete",
playlistId,
entryId,
});
const result = removeEntryTransaction(playlistId, entryId);
const playback = result.playbackChanged ? readPlayback.get() : null;
notify({
reason: "entry_deleted",
playlistId,
revision: result.playlist.revision,
entryId,
...(playback ? { playbackRevision: playback.revision } : {}),
});
return { id: entryId, deleted: true };
}
@@ -903,6 +855,7 @@ function createPlaylistService({
});
function reorderEntries(playlistId, orderedEntryIds) {
onBeforeChange({ reason: "before_entries_reorder", playlistId });
const result = reorderTransaction(playlistId, orderedEntryIds);
if (result.changed) {
notify({ reason: "entries_reordered", playlistId, revision: result.playlist.revision });
@@ -910,130 +863,7 @@ function createPlaylistService({
return result.detail;
}
function isCurrentNormalSlideMidDuration(state, nowMs) {
if (state.static_asset_id !== null || state.current_entry_id === null) return false;
if (state.active_playlist_id === null) return false;
const current = findCurrentEntry.get(
state.current_entry_id,
state.active_playlist_id,
);
if (!current || !evaluateEntryEligibility(current, current, nowMs).playable) {
return false;
}
if (state.rotation_paused) {
return (
Number.isSafeInteger(state.paused_remaining_ms) &&
state.paused_remaining_ms > 0
);
}
return Number.isSafeInteger(state.next_slide_at) && state.next_slide_at > nowMs;
}
function prepareImmediateTarget(playlistId, state, nowMs) {
const settings = readSettings.get();
if (!settings) throw new Error("The app_settings singleton row is missing");
const target = resolvedEntries
.all(playlistId)
.find((row) => evaluateEntryEligibility(row, row, nowMs).playable);
if (!target) {
return {
currentEntryId: null,
pausedRemainingMs: null,
slideStartedAt: null,
nextSlideAt: null,
};
}
const durationMs = resolveEntrySettings(target, settings).durationMs;
if (state.static_asset_id !== null || state.rotation_paused) {
return {
currentEntryId: target.entry_id,
pausedRemainingMs: state.rotation_paused ? durationMs : null,
slideStartedAt: null,
nextSlideAt: null,
};
}
return {
currentEntryId: target.entry_id,
pausedRemainingMs: null,
slideStartedAt: nowMs,
nextSlideAt: nowMs + durationMs,
};
}
const activateTransaction = database.transaction((playlistId) => {
const playlist = getPlaylistRow(playlistId);
const state = readPlayback.get();
if (!state) throw new Error("The global playback_state row is missing");
const requestedAt = now();
if (state.active_playlist_id === playlistId) {
if (state.pending_playlist_id === null) {
return { changed: false, activation: "already_active", playlist, state };
}
const timestamp = Math.max(requestedAt, state.updated_at + 1);
clearPendingActivation.run(timestamp);
return {
changed: true,
activation: "already_active",
playlist,
state: readPlayback.get(),
};
}
if (
state.pending_playlist_id === playlistId &&
isCurrentNormalSlideMidDuration(state, requestedAt)
) {
return { changed: false, activation: "pending", playlist, state };
}
const timestamp = Math.max(requestedAt, state.updated_at + 1);
if (isCurrentNormalSlideMidDuration(state, requestedAt)) {
setPendingActivation.run(playlistId, requestedAt, timestamp);
return {
changed: true,
activation: "pending",
playlist,
state: readPlayback.get(),
};
}
activateImmediately.run({
playlistId,
...prepareImmediateTarget(playlistId, state, requestedAt),
updatedAt: timestamp,
});
return {
changed: true,
activation: "active",
playlist,
state: readPlayback.get(),
};
});
function activate(playlistId) {
const result = activateTransaction(playlistId);
const playback = mapPlaybackState(result.state);
if (result.changed) {
notify({
reason:
result.activation === "pending"
? "activation_requested"
: "activated",
playlistId,
revision: result.playlist.revision,
playbackRevision: playback.revision,
});
}
return {
activation: result.activation,
playlistId,
playback,
};
}
return {
activate,
addEntry,
create,
get,
@@ -1050,6 +880,5 @@ module.exports = {
MAX_DURATION_MS,
MIN_DURATION_MS,
createPlaylistService,
mapPlaybackState,
normalizePlaylistName,
};
+37 -2
View File
@@ -4,6 +4,7 @@ const { createApp } = require("./app");
const { openDatabase } = require("./database");
const { runMigrations } = require("./database/migrate");
const { createImageProcessor } = require("./image-processing");
const { createPlaybackService } = require("./playback");
const { createStorageService } = require("./storage");
function listen(httpServer, port, host) {
@@ -54,7 +55,29 @@ function createRuntime({
}
let app;
let io = null;
let playbackService;
function emitSocketEvent(eventName, payload) {
try {
io?.emit(eventName, payload);
} catch (error) {
try {
logger.warn?.(`Socket.IO ${eventName} publish failed`, error);
} catch {
// Socket invalidation remains best effort after the authoritative commit.
}
}
}
try {
playbackService = createPlaybackService({
database,
logger,
onActivationChanged(event) {
emitSocketEvent("playlists:changed", event);
},
onStateChanged(snapshot) {
emitSocketEvent("playback:state", snapshot);
},
});
app = createApp({
config,
database,
@@ -63,8 +86,9 @@ function createRuntime({
logger,
migrations,
onPlaylistsChanged(event) {
io?.emit("playlists:changed", event);
emitSocketEvent("playlists:changed", event);
},
playbackService,
storage,
});
} catch (error) {
@@ -79,6 +103,11 @@ function createRuntime({
io.on("connection", (socket) => {
logger.info?.(`Socket.IO client connected: ${socket.id}`);
try {
socket.emit("playback:state", playbackService.getSnapshot());
} catch (error) {
logger.warn?.("Initial playback snapshot could not be sent", error);
}
socket.on("disconnect", () => {
logger.info?.(`Socket.IO client disconnected: ${socket.id}`);
});
@@ -96,7 +125,10 @@ function createRuntime({
return startPromise;
}
startPromise = listen(httpServer, config.port, config.host)
startPromise = Promise.resolve()
.then(() => playbackService.initialize())
.then(() => playbackService.start())
.then(() => listen(httpServer, config.port, config.host))
.then((listeningAddress) => {
address = listeningAddress;
lifecycle.ready = true;
@@ -108,6 +140,7 @@ function createRuntime({
return listeningAddress;
})
.catch((error) => {
playbackService.stop();
storage.stop();
database.close();
throw error;
@@ -122,6 +155,7 @@ function createRuntime({
lifecycle.ready = false;
lifecycle.stopping = true;
playbackService.stop();
stopPromise = new Promise((resolve) => {
let settled = false;
const finish = () => {
@@ -163,6 +197,7 @@ function createRuntime({
io,
lifecycle,
migrations,
playbackService,
storage,
imageProcessor,
get address() {
+10 -3
View File
@@ -2,7 +2,11 @@ const express = require("express");
function createSettingsRouter(
settingsService,
{ onPlaylistsChanged = () => {} } = {},
{
onBeforePlaybackDomainChange = () => {},
onPlaylistsChanged = () => {},
onPlaybackDomainChanged = () => {},
} = {},
) {
const router = express.Router();
@@ -13,13 +17,16 @@ function createSettingsRouter(
router.patch("/", (request, response, next) => {
try {
const previous = settingsService.get();
onBeforePlaybackDomainChange({ reason: "before_settings_update" });
const updated = settingsService.update(request.body);
if (updated.revision !== previous.revision) {
onPlaylistsChanged({
const change = {
reason: "settings_updated",
playlistIds: null,
settingsRevision: updated.revision,
});
};
onPlaybackDomainChanged(change);
onPlaylistsChanged(change);
}
response.json(updated);
} catch (error) {
+249
View File
@@ -0,0 +1,249 @@
const test = require("node:test");
const assert = require("node:assert/strict");
const { createApp } = require("../src/app");
const { loadConfig } = require("../src/config");
const { openDatabase } = require("../src/database");
const { runMigrations } = require("../src/database/migrate");
function testConfig() {
return loadConfig({
NODE_ENV: "test",
HOST: "127.0.0.1",
PORT: "0",
DATABASE_PATH: ":memory:",
APP_TIMEZONE: "America/Chicago",
S3_ENDPOINT: "http://seaweedfs:8333",
S3_BUCKET: "announcements-test",
});
}
function createPlaybackStub({ domainError = null } = {}) {
const calls = [];
let revision = 10;
const snapshot = (action = "get") => ({
protocolVersion: 1,
outputId: "global",
playbackRevision: revision,
action,
});
const action = (name) => () => {
calls.push([name]);
revision += 1;
return snapshot(name);
};
return {
calls,
domainChanged(change) {
calls.push(["domainChanged", change]);
if (domainError) throw domainError;
return snapshot("domainChanged");
},
getSnapshot() {
calls.push(["getSnapshot"]);
return snapshot();
},
pause: action("pause"),
reconcile(reason) {
calls.push(["reconcile", reason]);
return snapshot("reconcile");
},
resume: action("resume"),
next: action("next"),
previous: action("previous"),
reload: action("reload"),
enterStatic(assetId) {
calls.push(["enterStatic", assetId]);
revision += 1;
return snapshot("static");
},
setMode(mode) {
calls.push(["setMode", mode]);
revision += 1;
return snapshot("mode");
},
activatePlaylist(playlistId) {
calls.push(["activatePlaylist", playlistId]);
revision += 1;
const current = snapshot("activate");
return {
activation: "active",
playlistId,
playback: { revision: current.playbackRevision },
snapshot: current,
};
},
};
}
async function createHarness(t, { domainError = null } = {}) {
const database = openDatabase({ path: ":memory:" });
const migrations = runMigrations(database);
const playbackService = createPlaybackStub({ domainError });
const invalidations = [];
const app = createApp({
config: testConfig(),
database,
imageProcessor: {},
lifecycle: { ready: true, stopping: false },
logger: { info() {}, warn() {}, error() {} },
migrations,
onPlaylistsChanged: (event) => invalidations.push(event),
playbackService,
storage: {
health() {
return { status: "healthy", checkedAt: null, lastSuccessAt: null };
},
},
});
const server = app.listen(0, "127.0.0.1");
await new Promise((resolve, reject) => {
server.once("listening", resolve);
server.once("error", reject);
});
t.after(async () => {
await new Promise((resolve) => server.close(resolve));
database.close();
});
return {
baseUrl: `http://127.0.0.1:${server.address().port}`,
database,
invalidations,
playbackService,
};
}
async function requestJson(baseUrl, path, { body, ...init } = {}) {
const response = await fetch(`${baseUrl}${path}`, {
...init,
...(body === undefined
? {}
: {
headers: { "content-type": "application/json", ...init.headers },
body: JSON.stringify(body),
}),
});
return { response, body: await response.json() };
}
test("playback REST routes delegate every command to the one playback service", async (t) => {
const harness = await createHarness(t);
const read = await requestJson(harness.baseUrl, "/api/playback");
assert.equal(read.response.status, 200);
assert.equal(read.body.playbackRevision, 10);
const commands = [
["POST", "/api/playback/pause", undefined, "pause"],
["POST", "/api/playback/resume", undefined, "resume"],
["POST", "/api/playback/next", undefined, "next"],
["POST", "/api/playback/previous", undefined, "previous"],
["POST", "/api/playback/reload", undefined, "reload"],
["POST", "/api/playback/static", { assetId: "asset-static" }, "static"],
["PUT", "/api/playback/mode", { mode: "off" }, "mode"],
];
for (const [method, path, body, action] of commands) {
const result = await requestJson(harness.baseUrl, path, { method, body });
assert.equal(result.response.status, 200);
assert.equal(result.body.action, action);
}
assert.deepEqual(harness.playbackService.calls, [
["getSnapshot"],
["pause"],
["resume"],
["next"],
["previous"],
["reload"],
["enterStatic", "asset-static"],
["setMode", "off"],
]);
});
test("playlist activation delegates directly to the playback authority", async (t) => {
const harness = await createHarness(t);
const result = await requestJson(
harness.baseUrl,
"/api/playlists/playlist-a/activate",
{ method: "POST" },
);
assert.equal(result.response.status, 200);
assert.equal(result.body.activation, "active");
assert.equal(result.body.playlistId, "playlist-a");
assert.deepEqual(harness.playbackService.calls, [
["activatePlaylist", "playlist-a"],
]);
assert.deepEqual(harness.invalidations, []);
});
test("semantic domain edits reconcile playback independently of list invalidation", async (t) => {
const harness = await createHarness(t);
const timestamp = Date.now();
harness.database
.prepare(
`INSERT INTO assets (
id, source_sha256, display_name, original_filename, status,
created_at, updated_at
) VALUES (?, ?, ?, ?, 'ready', ?, ?)`,
)
.run(
"unreferenced-asset",
"a".repeat(64),
"Unreferenced",
"unreferenced.png",
timestamp,
timestamp,
);
const settings = await requestJson(harness.baseUrl, "/api/settings", {
method: "PATCH",
body: { defaultBlurPx: 21 },
});
assert.equal(settings.response.status, 200);
const asset = await requestJson(
harness.baseUrl,
"/api/assets/unreferenced-asset",
{ method: "PATCH", body: { displayName: "Renamed unreferenced" } },
);
assert.equal(asset.response.status, 200);
assert.deepEqual(harness.playbackService.calls, [
["reconcile", "before_domain_change"],
[
"domainChanged",
{ reason: "settings_updated", playlistIds: null, settingsRevision: 2 },
],
["reconcile", "before_domain_change"],
[
"domainChanged",
{
reason: "asset_updated",
playlistIds: [],
assetId: "unreferenced-asset",
},
],
]);
assert.deepEqual(harness.invalidations, [
{ reason: "settings_updated", playlistIds: null, settingsRevision: 2 },
]);
});
test("a playback reconciliation failure is surfaced after the domain commit", async (t) => {
const harness = await createHarness(t, {
domainError: new Error("reconciliation failed"),
});
const result = await requestJson(harness.baseUrl, "/api/playlists", {
method: "POST",
body: { name: "Committed before reconciliation" },
});
assert.equal(result.response.status, 500);
assert.equal(result.body.error.code, "INTERNAL_ERROR");
assert.equal(
harness.database.prepare("SELECT COUNT(*) AS count FROM playlists").get()
.count,
1,
);
assert.equal(harness.invalidations.length, 0);
assert.equal(harness.playbackService.calls[0][0], "domainChanged");
});
File diff suppressed because it is too large Load Diff
+35 -17
View File
@@ -4,6 +4,7 @@ const crypto = require("node:crypto");
const { openDatabase } = require("../src/database");
const { runMigrations } = require("../src/database/migrate");
const { createAssetService } = require("../src/assets/service");
const { createPlaybackService } = require("../src/playback");
const {
MAX_DURATION_MS,
MIN_DURATION_MS,
@@ -19,13 +20,22 @@ function createHarness() {
let idCounter = 0;
let assetCounter = 0;
const changes = [];
let playbackService;
const service = createPlaylistService({
database,
now: () => clock,
createId: () =>
`00000000-0000-4000-8000-${String(++idCounter).padStart(12, "0")}`,
onChanged: (change) => changes.push(change),
onBeforeChange: () => {
playbackService?.reconcile("before_domain_change");
},
onChanged: (change) => {
changes.push(change);
playbackService?.domainChanged(change);
},
});
playbackService = createPlaybackService({ database, now: () => clock });
playbackService.initialize();
function setNow(value) {
clock = value;
@@ -71,7 +81,15 @@ function createHarness() {
.get();
}
return { changes, database, playback, seedAsset, service, setNow };
return {
changes,
database,
playback,
playbackService,
seedAsset,
service,
setNow,
};
}
function expectError(action, code, statusCode) {
@@ -479,7 +497,7 @@ test("activation queues at a live boundary, is idempotent, and mode off is only
durationOverrideMs: 4000,
});
const initial = harness.service.activate(oldPlaylist.id);
const initial = harness.playbackService.activatePlaylist(oldPlaylist.id);
assert.equal(initial.activation, "active");
assert.equal(initial.playback.currentEntryId, oldEntry.id);
assert.equal(initial.playback.slideStartedAt, BASE_TIME);
@@ -489,7 +507,7 @@ test("activation queues at a live boundary, is idempotent, and mode off is only
.prepare("UPDATE playback_state SET mode = 'off' WHERE output_id = 'global'")
.run();
harness.setNow(BASE_TIME + 1000);
const pending = harness.service.activate(newPlaylist.id);
const pending = harness.playbackService.activatePlaylist(newPlaylist.id);
assert.equal(pending.activation, "pending");
assert.equal(pending.playback.mode, "off");
assert.equal(pending.playback.activePlaylistId, oldPlaylist.id);
@@ -498,18 +516,18 @@ test("activation queues at a live boundary, is idempotent, and mode off is only
const pendingRevision = pending.playback.revision;
const requestedAt = pending.playback.pendingPlaylistRequestedAt;
const repeated = harness.service.activate(newPlaylist.id);
const repeated = harness.playbackService.activatePlaylist(newPlaylist.id);
assert.equal(repeated.activation, "pending");
assert.equal(repeated.playback.revision, pendingRevision);
assert.equal(repeated.playback.pendingPlaylistRequestedAt, requestedAt);
const cancel = harness.service.activate(oldPlaylist.id);
const cancel = harness.playbackService.activatePlaylist(oldPlaylist.id);
assert.equal(cancel.activation, "already_active");
assert.equal(cancel.playback.pendingPlaylistId, null);
assert.equal(cancel.playback.currentEntryId, oldEntry.id);
harness.setNow(BASE_TIME + 7000);
const immediate = harness.service.activate(newPlaylist.id);
const immediate = harness.playbackService.activatePlaylist(newPlaylist.id);
assert.equal(immediate.activation, "active");
assert.equal(immediate.playback.activePlaylistId, newPlaylist.id);
assert.equal(immediate.playback.pendingPlaylistId, null);
@@ -534,7 +552,7 @@ test("paused normal playback queues and can replace a pending activation", (t) =
harness.service.addEntry(secondTarget.id, {
assetId: harness.seedAsset({ id: "second-target", name: "Second target" }),
});
harness.service.activate(current.id);
harness.playbackService.activatePlaylist(current.id);
harness.database
.prepare(
`UPDATE playback_state
@@ -545,14 +563,14 @@ test("paused normal playback queues and can replace a pending activation", (t) =
.run();
harness.setNow(BASE_TIME + 1000);
const first = harness.service.activate(firstTarget.id);
const first = harness.playbackService.activatePlaylist(firstTarget.id);
assert.equal(first.activation, "pending");
assert.equal(first.playback.activePlaylistId, current.id);
assert.equal(first.playback.pendingPlaylistId, firstTarget.id);
assert.equal(first.playback.pausedRemainingMs, 5000);
harness.setNow(BASE_TIME + 2000);
const replacement = harness.service.activate(secondTarget.id);
const replacement = harness.playbackService.activatePlaylist(secondTarget.id);
assert.equal(replacement.activation, "pending");
assert.equal(replacement.playback.activePlaylistId, current.id);
assert.equal(replacement.playback.pendingPlaylistId, secondTarget.id);
@@ -579,7 +597,7 @@ test("static and paused activation prepare the first playable normal target with
durationOverrideMs: 9000,
});
assert.equal(skipped.position, 0);
harness.service.activate(oldPlaylist.id);
harness.playbackService.activatePlaylist(oldPlaylist.id);
harness.database
.prepare(
`UPDATE playback_state
@@ -590,7 +608,7 @@ test("static and paused activation prepare the first playable normal target with
)
.run(staticAssetId);
const result = harness.service.activate(target.id);
const result = harness.playbackService.activatePlaylist(target.id);
assert.equal(result.activation, "active");
assert.equal(result.playback.activePlaylistId, target.id);
assert.equal(result.playback.staticAssetId, staticAssetId);
@@ -611,7 +629,7 @@ test("empty activation is immediate and current-entry deletion coordinates playb
assetId: harness.seedAsset({ id: `asset-${id}`, name: id }),
}),
);
harness.service.activate(populated.id);
harness.playbackService.activatePlaylist(populated.id);
const futurePlaybackTimestamp = BASE_TIME + 100_000;
harness.database
.prepare(
@@ -624,10 +642,10 @@ test("empty activation is immediate and current-entry deletion coordinates playb
harness.service.removeEntry(populated.id, entries[0].id);
const afterDelete = harness.playback();
assert.equal(afterDelete.current_entry_id, null);
assert.equal(afterDelete.current_entry_id, entries[1].id);
assert.equal(afterDelete.slide_started_at, null);
assert.equal(afterDelete.next_slide_at, null);
assert.equal(afterDelete.paused_remaining_ms, null);
assert.equal(afterDelete.paused_remaining_ms, 7000);
assert.equal(afterDelete.active_playlist_id, populated.id);
assert.equal(afterDelete.mode, "off");
assert.equal(afterDelete.static_asset_id, "asset-second");
@@ -646,7 +664,7 @@ test("empty activation is immediate and current-entry deletion coordinates playb
WHERE output_id = 'global'`,
)
.run();
const result = harness.service.activate(empty.id);
const result = harness.playbackService.activatePlaylist(empty.id);
assert.equal(result.activation, "active");
assert.equal(result.playback.activePlaylistId, empty.id);
assert.equal(result.playback.currentEntryId, null);
@@ -665,7 +683,7 @@ test("inactive playlist edits do not mutate authoritative playback state", (t) =
const draftEntry = harness.service.addEntry(inactive.id, {
assetId: harness.seedAsset({ id: "draft", name: "Draft asset" }),
});
harness.service.activate(active.id);
harness.playbackService.activatePlaylist(active.id);
const before = harness.playback();
harness.service.update(inactive.id, { name: "Draft edited" });
+28 -1
View File
@@ -1,9 +1,10 @@
const test = require("node:test");
const assert = require("node:assert/strict");
const net = require("node:net");
const { loadConfig } = require("../src/config");
const { createRuntime } = require("../src/runtime");
function testConfig() {
function testConfig(overrides = {}) {
return loadConfig({
NODE_ENV: "test",
HOST: "127.0.0.1",
@@ -20,6 +21,7 @@ function testConfig() {
MQTT_URL: "mqtt://broker:1883",
MQTT_PASSWORD: "mqtt-password-that-must-not-leak",
HA_WEBHOOK_URL: "http://home-assistant:8123/api/webhook/private-id",
...overrides,
});
}
@@ -171,3 +173,28 @@ test("runtime startup and shutdown are idempotent", async () => {
assert.equal(runtime.lifecycle.ready, false);
assert.equal(runtime.database.open, false);
});
test("a listen failure stops playback and closes startup resources", async (t) => {
const occupied = net.createServer();
occupied.listen(0, "127.0.0.1");
await new Promise((resolve, reject) => {
occupied.once("listening", resolve);
occupied.once("error", reject);
});
t.after(
() =>
new Promise((resolve) => {
occupied.close(resolve);
}),
);
const runtime = createRuntime({
config: testConfig({ PORT: String(occupied.address().port) }),
logger: silentLogger,
storage: testStorage(),
});
await assert.rejects(runtime.start(), (error) => error.code === "EADDRINUSE");
assert.equal(runtime.lifecycle.ready, false);
assert.equal(runtime.database.open, false);
});
+185 -12
View File
@@ -97,6 +97,7 @@ async function connectPollingSocket(baseUrl) {
);
return {
initialEvents: socketEvents(connectedPayload),
async poll() {
const response = await fetch(endpoint(sid), {
signal: AbortSignal.timeout(3_000),
@@ -115,6 +116,65 @@ async function connectPollingSocket(baseUrl) {
};
}
test("Socket.IO sends an authoritative playback snapshot on connect and after mutations", async (t) => {
const runtime = createRuntime({
config: testConfig(),
logger: silentLogger,
storage: testStorage(),
});
await runtime.start();
t.after(() => runtime.stop("test cleanup"));
const baseUrl = `http://127.0.0.1:${runtime.address.port}`;
const socket = await connectPollingSocket(baseUrl);
t.after(() => socket.close().catch(() => {}));
let initial = socket.initialEvents.find(
([eventName]) => eventName === "playback:state",
);
if (!initial) {
initial = socketEvents(await socket.poll()).find(
([eventName]) => eventName === "playback:state",
);
}
assert.ok(initial, "a full playback snapshot must be sent on connection");
assert.equal(initial[1].protocolVersion, 1);
assert.equal(initial[1].outputId, "global");
const initialRevision = initial[1].playbackRevision;
const read = await jsonRequest(baseUrl, "/api/playback");
assert.equal(read.response.status, 200);
assert.equal(read.body.playbackRevision, initialRevision);
const reread = await jsonRequest(baseUrl, "/api/playback");
assert.equal(reread.body.playbackRevision, initialRevision);
const changed = await jsonRequest(baseUrl, "/api/playback/mode", {
method: "PUT",
body: { mode: "off" },
});
assert.equal(changed.response.status, 200);
assert.equal(changed.body.mode, "off");
assert.equal(changed.body.playbackRevision, initialRevision + 1);
const events = socketEvents(await socket.poll());
const playbackEvents = events.filter(
([eventName]) => eventName === "playback:state",
);
assert.equal(playbackEvents.length, 1);
assert.deepEqual(playbackEvents[0][1], changed.body);
const unchanged = await jsonRequest(baseUrl, "/api/playback/mode", {
method: "PUT",
body: { mode: "off" },
});
assert.equal(unchanged.body.playbackRevision, changed.body.playbackRevision);
assert.equal(
runtime.database
.prepare("SELECT revision FROM playback_state WHERE output_id = 'global'")
.get().revision,
changed.body.playbackRevision,
);
});
test("Socket.IO publishes playlists:changed only after successful semantic mutations", async (t) => {
const runtime = createRuntime({
config: testConfig(),
@@ -144,10 +204,12 @@ test("Socket.IO publishes playlists:changed only after successful semantic mutat
});
assert.equal(created.response.status, 201);
let events = socketEvents(await socket.poll());
assert.equal(events.length, 1);
assert.equal(events[0][0], "playlists:changed");
assert.equal(events[0][1].reason, "created");
assert.equal(events[0][1].playlistId, created.body.id);
let playlistEvents = events.filter(
([eventName]) => eventName === "playlists:changed",
);
assert.equal(playlistEvents.length, 1);
assert.equal(playlistEvents[0][1].reason, "created");
assert.equal(playlistEvents[0][1].playlistId, created.body.id);
const duplicate = await jsonRequest(baseUrl, "/api/playlists", {
method: "POST",
@@ -163,7 +225,10 @@ test("Socket.IO publishes playlists:changed only after successful semantic mutat
body: { defaultDurationMs: 8_000 },
});
events = socketEvents(await socket.poll());
assert.deepEqual(events, [
playlistEvents = events.filter(
([eventName]) => eventName === "playlists:changed",
);
assert.deepEqual(playlistEvents, [
[
"playlists:changed",
{
@@ -180,8 +245,11 @@ test("Socket.IO publishes playlists:changed only after successful semantic mutat
});
assert.equal(second.response.status, 201);
events = socketEvents(await socket.poll());
assert.equal(events.length, 1);
assert.equal(events[0][1].reason, "created");
playlistEvents = events.filter(
([eventName]) => eventName === "playlists:changed",
);
assert.equal(playlistEvents.length, 1);
assert.equal(playlistEvents[0][1].reason, "created");
const activated = await jsonRequest(
baseUrl,
@@ -191,14 +259,20 @@ test("Socket.IO publishes playlists:changed only after successful semantic mutat
assert.equal(activated.response.status, 200);
assert.equal(activated.body.activation, "active");
events = socketEvents(await socket.poll());
assert.equal(events.length, 1);
assert.equal(events[0][0], "playlists:changed");
assert.equal(events[0][1].reason, "activated");
assert.equal(events[0][1].playlistId, created.body.id);
playlistEvents = events.filter(
([eventName]) => eventName === "playlists:changed",
);
assert.equal(playlistEvents.length, 1);
assert.equal(playlistEvents[0][1].reason, "activated");
assert.equal(playlistEvents[0][1].playlistId, created.body.id);
assert.equal(
events[0][1].playbackRevision,
playlistEvents[0][1].playbackRevision,
activated.body.playback.revision,
);
assert.equal(
events.filter(([eventName]) => eventName === "playback:state").length,
1,
);
const activeDelete = await jsonRequest(
baseUrl,
@@ -208,3 +282,102 @@ test("Socket.IO publishes playlists:changed only after successful semantic mutat
assert.equal(activeDelete.response.status, 409);
assert.equal(activeDelete.body.error.code, "PLAYLIST_ACTIVE");
});
test("activation invalidations come from authoritative pending-state transitions", async (t) => {
const runtime = createRuntime({
config: testConfig(),
logger: silentLogger,
storage: testStorage(),
});
await runtime.start();
t.after(() => runtime.stop("test cleanup"));
const timestamp = Date.now();
const insertAsset = runtime.database.prepare(`
INSERT INTO assets (
id, source_sha256, display_name, original_filename, status,
display_sha256, display_content_type, display_byte_size,
display_width, display_height, created_at, updated_at
) VALUES (?, ?, ?, ?, 'ready', ?, 'image/webp', 1, 1920, 1080, ?, ?)
`);
const insertPlaylist = runtime.database.prepare(`
INSERT INTO playlists (id, name, created_at, updated_at)
VALUES (?, ?, ?, ?)
`);
const insertEntry = runtime.database.prepare(`
INSERT INTO playlist_entries (
id, playlist_id, asset_id, position, created_at, updated_at
) VALUES (?, ?, ?, 0, ?, ?)
`);
insertAsset.run(
"asset-old",
"a".repeat(64),
"Old asset",
"old.png",
"c".repeat(64),
timestamp,
timestamp,
);
insertAsset.run(
"asset-new",
"b".repeat(64),
"New asset",
"new.png",
"d".repeat(64),
timestamp,
timestamp,
);
insertPlaylist.run("playlist-old", "Old", timestamp, timestamp);
insertPlaylist.run("playlist-new", "New", timestamp, timestamp);
insertEntry.run(
"entry-old",
"playlist-old",
"asset-old",
timestamp,
timestamp,
);
insertEntry.run(
"entry-new",
"playlist-new",
"asset-new",
timestamp,
timestamp,
);
const baseUrl = `http://127.0.0.1:${runtime.address.port}`;
const socket = await connectPollingSocket(baseUrl);
t.after(() => socket.close().catch(() => {}));
const activateOld = await jsonRequest(
baseUrl,
"/api/playlists/playlist-old/activate",
{ method: "POST" },
);
assert.equal(activateOld.body.activation, "active");
await socket.poll();
const requestNew = await jsonRequest(
baseUrl,
"/api/playlists/playlist-new/activate",
{ method: "POST" },
);
assert.equal(requestNew.body.activation, "pending");
let invalidations = socketEvents(await socket.poll()).filter(
([eventName]) => eventName === "playlists:changed",
);
assert.equal(invalidations.length, 1);
assert.equal(invalidations[0][1].reason, "activation_requested");
assert.equal(invalidations[0][1].pendingPlaylistId, "playlist-new");
const cancel = await jsonRequest(
baseUrl,
"/api/playlists/playlist-old/activate",
{ method: "POST" },
);
assert.equal(cancel.body.activation, "already_active");
invalidations = socketEvents(await socket.poll()).filter(
([eventName]) => eventName === "playlists:changed",
);
assert.equal(invalidations.length, 1);
assert.equal(invalidations[0][1].reason, "activation_canceled");
assert.equal(invalidations[0][1].pendingPlaylistId, null);
});