feat: define presentation sync protocol
This commit is contained in:
+1
-1
@@ -9,7 +9,7 @@
|
||||
"dev": "concurrently --kill-others-on-fail --names server,console --prefix-colors blue,green \"pnpm --filter @lda/web-server dev\" \"pnpm --filter @lda/console dev\"",
|
||||
"test": "pnpm -r --if-present test",
|
||||
"typecheck": "pnpm -r --if-present typecheck",
|
||||
"build": "pnpm --filter @lda/workflow-rpc build && pnpm --filter @lda/console build && pnpm --filter @lda/web-server build",
|
||||
"build": "pnpm --filter @lda/workflow-rpc build && pnpm --filter @lda/presentation-sync build && pnpm --filter @lda/console build && pnpm --filter @lda/web-server build",
|
||||
"start": "pnpm --filter @lda/web-server start"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"name": "@lda/presentation-sync",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./dist/index.d.ts",
|
||||
"default": "./dist/index.js"
|
||||
}
|
||||
},
|
||||
"scripts": {
|
||||
"build": "tsc -p tsconfig.json",
|
||||
"test": "vitest run",
|
||||
"typecheck": "tsc -p tsconfig.json --noEmit"
|
||||
},
|
||||
"dependencies": {
|
||||
"effect": "3.21.4"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "26.1.0",
|
||||
"vitest": "4.1.9"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
export {
|
||||
JOIN_CODE_LENGTH,
|
||||
MAX_PRESENTATION_HASH_LENGTH,
|
||||
MAX_SYNC_MESSAGE_BYTES,
|
||||
decodeClientSyncMessage,
|
||||
decodeCreateSessionRequest,
|
||||
decodeJoinSessionRequest,
|
||||
decodeServerSyncMessage,
|
||||
isCanonicalPresentationHash,
|
||||
normalizeJoinCode,
|
||||
} from "./protocol.js";
|
||||
|
||||
export type {
|
||||
ClientSyncMessage,
|
||||
CreateSessionRequest,
|
||||
DecodeResult,
|
||||
JoinSessionRequest,
|
||||
PresentationPresence,
|
||||
PresentationRole,
|
||||
PresentationSnapshot,
|
||||
ServerSyncMessage,
|
||||
SessionGrant,
|
||||
} from "./protocol.js";
|
||||
@@ -0,0 +1,240 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
JOIN_CODE_LENGTH,
|
||||
MAX_PRESENTATION_HASH_LENGTH,
|
||||
MAX_SYNC_MESSAGE_BYTES,
|
||||
decodeClientSyncMessage,
|
||||
decodeCreateSessionRequest,
|
||||
decodeJoinSessionRequest,
|
||||
decodeServerSyncMessage,
|
||||
isCanonicalPresentationHash,
|
||||
normalizeJoinCode,
|
||||
} from "./protocol.js";
|
||||
|
||||
describe("presentation sync protocol", () => {
|
||||
it("accepts bounded canonical location publishes", () => {
|
||||
expect(
|
||||
decodeClientSyncMessage(
|
||||
JSON.stringify({
|
||||
type: "location.publish",
|
||||
hash: "#scene/architecture/client/focus/client-operations",
|
||||
baseRevision: 4,
|
||||
messageId: "msg-4",
|
||||
}),
|
||||
),
|
||||
).toEqual({
|
||||
ok: true,
|
||||
value: {
|
||||
type: "location.publish",
|
||||
hash: "#scene/architecture/client/focus/client-operations",
|
||||
baseRevision: 4,
|
||||
messageId: "msg-4",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["", "#unknown/x", "scene/thesis/title"])(
|
||||
"rejects non-canonical hash %s",
|
||||
(hash) => expect(isCanonicalPresentationHash(hash)).toBe(false),
|
||||
);
|
||||
|
||||
it("enforces the canonical hash length bound", () => {
|
||||
expect(
|
||||
isCanonicalPresentationHash(
|
||||
`#scene/${"x".repeat(MAX_PRESENTATION_HASH_LENGTH)}`,
|
||||
),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it("rejects oversized websocket payloads before JSON decoding", () => {
|
||||
const result = decodeClientSyncMessage(
|
||||
"x".repeat(MAX_SYNC_MESSAGE_BYTES + 1),
|
||||
);
|
||||
expect(result).toEqual({ ok: false, error: "message_too_large" });
|
||||
});
|
||||
|
||||
it("normalizes human-entered join codes", () => {
|
||||
expect(normalizeJoinCode(" ab-cd 7 ")).toBe("ABCD7");
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ type: "location.unknown" },
|
||||
{
|
||||
type: "location.publish",
|
||||
hash: "#scene/thesis/title",
|
||||
baseRevision: -1,
|
||||
messageId: "msg-1",
|
||||
},
|
||||
{
|
||||
type: "location.publish",
|
||||
hash: "#scene/thesis/title",
|
||||
baseRevision: 1.5,
|
||||
messageId: "msg-1",
|
||||
},
|
||||
{
|
||||
type: "location.publish",
|
||||
hash: "#scene/thesis/title",
|
||||
baseRevision: 0,
|
||||
messageId: "x".repeat(129),
|
||||
},
|
||||
{
|
||||
type: "location.publish",
|
||||
hash: `#scene/${"x".repeat(MAX_PRESENTATION_HASH_LENGTH)}`,
|
||||
baseRevision: 0,
|
||||
messageId: "msg-1",
|
||||
},
|
||||
])("rejects invalid client message %#", (message) => {
|
||||
expect(decodeClientSyncMessage(JSON.stringify(message))).toEqual({
|
||||
ok: false,
|
||||
error: "invalid_message",
|
||||
});
|
||||
});
|
||||
|
||||
it("accepts session end and ping messages", () => {
|
||||
expect(decodeClientSyncMessage('{"type":"session.end"}')).toEqual({
|
||||
ok: true,
|
||||
value: { type: "session.end" },
|
||||
});
|
||||
expect(
|
||||
decodeClientSyncMessage(JSON.stringify({ type: "ping", nonce: "n-1" })),
|
||||
).toEqual({
|
||||
ok: true,
|
||||
value: { type: "ping", nonce: "n-1" },
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
role: "presenter",
|
||||
initialHash: "#scene/thesis/title",
|
||||
},
|
||||
{
|
||||
role: "audience",
|
||||
initialHash: "#discuss/problem/direct-actions",
|
||||
},
|
||||
])("accepts a valid create request %#", (request) => {
|
||||
expect(decodeCreateSessionRequest(JSON.stringify(request))).toEqual({
|
||||
ok: true,
|
||||
value: request,
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects invalid create request roles and hashes", () => {
|
||||
expect(
|
||||
decodeCreateSessionRequest(
|
||||
JSON.stringify({ role: "moderator", initialHash: "#scene/thesis/title" }),
|
||||
),
|
||||
).toEqual({ ok: false, error: "invalid_message" });
|
||||
expect(
|
||||
decodeCreateSessionRequest(
|
||||
JSON.stringify({ role: "presenter", initialHash: "#unknown/title" }),
|
||||
),
|
||||
).toEqual({ ok: false, error: "invalid_message" });
|
||||
});
|
||||
|
||||
it("normalizes and validates join request codes", () => {
|
||||
expect(
|
||||
decodeJoinSessionRequest(
|
||||
JSON.stringify({ role: "audience", code: " ab-cd 7x " }),
|
||||
),
|
||||
).toEqual({
|
||||
ok: true,
|
||||
value: { role: "audience", code: "ABCD7X" },
|
||||
});
|
||||
expect(
|
||||
decodeJoinSessionRequest(
|
||||
JSON.stringify({ role: "audience", code: "AB12" }),
|
||||
),
|
||||
).toEqual({ ok: false, error: "invalid_message" });
|
||||
expect(
|
||||
decodeJoinSessionRequest(
|
||||
JSON.stringify({ role: "moderator", code: "ABCD7X" }),
|
||||
),
|
||||
).toEqual({ ok: false, error: "invalid_message" });
|
||||
expect(JOIN_CODE_LENGTH).toBe(6);
|
||||
});
|
||||
|
||||
it("decodes every server message variant", () => {
|
||||
const messages = [
|
||||
{
|
||||
type: "location.snapshot",
|
||||
snapshot: { hash: "#scene/thesis/title", revision: 3 },
|
||||
originatingMessageId: "msg-3",
|
||||
},
|
||||
{
|
||||
type: "location.snapshot",
|
||||
snapshot: { hash: "#discuss/problem/direct-actions", revision: 0 },
|
||||
originatingMessageId: null,
|
||||
},
|
||||
{
|
||||
type: "presence.snapshot",
|
||||
presence: { presenters: 1, audience: 2 },
|
||||
},
|
||||
{
|
||||
type: "location.rejected",
|
||||
reason: "stale_revision",
|
||||
current: { hash: "#scene/thesis/title", revision: 4 },
|
||||
messageId: "msg-stale",
|
||||
},
|
||||
{ type: "session.ended", reason: "presenter_ended" },
|
||||
{ type: "session.ended", reason: "expired" },
|
||||
{
|
||||
type: "protocol.error",
|
||||
code: "invalid_message",
|
||||
message: "invalid payload",
|
||||
},
|
||||
{
|
||||
type: "protocol.error",
|
||||
code: "message_too_large",
|
||||
message: "payload exceeded limit",
|
||||
},
|
||||
{
|
||||
type: "protocol.error",
|
||||
code: "forbidden",
|
||||
message: "presenter role required",
|
||||
},
|
||||
];
|
||||
|
||||
for (const message of messages) {
|
||||
expect(decodeServerSyncMessage(JSON.stringify(message))).toEqual({
|
||||
ok: true,
|
||||
value: message,
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects invalid server revisions and presence counts", () => {
|
||||
expect(
|
||||
decodeServerSyncMessage(
|
||||
JSON.stringify({
|
||||
type: "location.snapshot",
|
||||
snapshot: { hash: "#scene/thesis/title", revision: 1.25 },
|
||||
originatingMessageId: null,
|
||||
}),
|
||||
),
|
||||
).toEqual({ ok: false, error: "invalid_message" });
|
||||
expect(
|
||||
decodeServerSyncMessage(
|
||||
JSON.stringify({
|
||||
type: "presence.snapshot",
|
||||
presence: { presenters: -1, audience: 0 },
|
||||
}),
|
||||
),
|
||||
).toEqual({ ok: false, error: "invalid_message" });
|
||||
});
|
||||
|
||||
it("returns stable errors for malformed JSON", () => {
|
||||
expect(decodeClientSyncMessage("not json")).toEqual({
|
||||
ok: false,
|
||||
error: "invalid_json",
|
||||
});
|
||||
expect(decodeServerSyncMessage("{")).toEqual({
|
||||
ok: false,
|
||||
error: "invalid_json",
|
||||
});
|
||||
expect(decodeCreateSessionRequest("[]")).toEqual({
|
||||
ok: false,
|
||||
error: "invalid_message",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,237 @@
|
||||
import { Schema } from "effect";
|
||||
|
||||
export const MAX_SYNC_MESSAGE_BYTES = 16 * 1024;
|
||||
export const MAX_PRESENTATION_HASH_LENGTH = 2_048;
|
||||
export const JOIN_CODE_LENGTH = 6;
|
||||
|
||||
const MAX_MESSAGE_ID_LENGTH = 128;
|
||||
|
||||
export type PresentationRole = "presenter" | "audience";
|
||||
|
||||
export type PresentationSnapshot = {
|
||||
readonly hash: string;
|
||||
readonly revision: number;
|
||||
};
|
||||
|
||||
export type PresentationPresence = {
|
||||
readonly presenters: number;
|
||||
readonly audience: number;
|
||||
};
|
||||
|
||||
export type CreateSessionRequest = {
|
||||
readonly role: PresentationRole;
|
||||
readonly initialHash: string;
|
||||
};
|
||||
|
||||
export type JoinSessionRequest = {
|
||||
readonly role: PresentationRole;
|
||||
readonly code: string;
|
||||
};
|
||||
|
||||
export type SessionGrant = {
|
||||
readonly sessionId: string;
|
||||
readonly code: string;
|
||||
readonly connectionToken: string;
|
||||
readonly websocketPath: "/api/presentation-sync/ws";
|
||||
readonly snapshot: PresentationSnapshot;
|
||||
};
|
||||
|
||||
export type ClientSyncMessage =
|
||||
| {
|
||||
readonly type: "location.publish";
|
||||
readonly hash: string;
|
||||
readonly baseRevision: number;
|
||||
readonly messageId: string;
|
||||
}
|
||||
| { readonly type: "session.end" }
|
||||
| { readonly type: "ping"; readonly nonce: string };
|
||||
|
||||
export type ServerSyncMessage =
|
||||
| {
|
||||
readonly type: "location.snapshot";
|
||||
readonly snapshot: PresentationSnapshot;
|
||||
readonly originatingMessageId: string | null;
|
||||
}
|
||||
| {
|
||||
readonly type: "presence.snapshot";
|
||||
readonly presence: PresentationPresence;
|
||||
}
|
||||
| {
|
||||
readonly type: "location.rejected";
|
||||
readonly reason: "stale_revision";
|
||||
readonly current: PresentationSnapshot;
|
||||
readonly messageId: string;
|
||||
}
|
||||
| {
|
||||
readonly type: "session.ended";
|
||||
readonly reason: "presenter_ended" | "expired";
|
||||
}
|
||||
| {
|
||||
readonly type: "protocol.error";
|
||||
readonly code: "invalid_message" | "message_too_large" | "forbidden";
|
||||
readonly message: string;
|
||||
};
|
||||
|
||||
export type DecodeResult<T> =
|
||||
| { readonly ok: true; readonly value: T }
|
||||
| {
|
||||
readonly ok: false;
|
||||
readonly error: "invalid_json" | "invalid_message" | "message_too_large";
|
||||
};
|
||||
|
||||
export const isCanonicalPresentationHash = (value: string): boolean =>
|
||||
value.length > 0 &&
|
||||
value.length <= MAX_PRESENTATION_HASH_LENGTH &&
|
||||
(value.startsWith("#scene/") || value.startsWith("#discuss/"));
|
||||
|
||||
export const normalizeJoinCode = (value: string): string =>
|
||||
value.replace(/[\s-]/g, "").toUpperCase();
|
||||
|
||||
const PresentationRoleSchema = Schema.Literal("presenter", "audience");
|
||||
const NonNegativeIntegerSchema = Schema.Number.pipe(
|
||||
Schema.int(),
|
||||
Schema.between(0, Number.MAX_SAFE_INTEGER),
|
||||
);
|
||||
const CanonicalHashSchema = Schema.String.pipe(
|
||||
Schema.filter(isCanonicalPresentationHash),
|
||||
);
|
||||
const BoundedMessageIdSchema = Schema.String.pipe(
|
||||
Schema.maxLength(MAX_MESSAGE_ID_LENGTH),
|
||||
);
|
||||
const SnapshotSchema = Schema.Struct({
|
||||
hash: CanonicalHashSchema,
|
||||
revision: NonNegativeIntegerSchema,
|
||||
});
|
||||
const PresenceSchema = Schema.Struct({
|
||||
presenters: NonNegativeIntegerSchema,
|
||||
audience: NonNegativeIntegerSchema,
|
||||
});
|
||||
|
||||
const ClientSyncMessageSchema = Schema.Union(
|
||||
Schema.Struct({
|
||||
type: Schema.Literal("location.publish"),
|
||||
hash: CanonicalHashSchema,
|
||||
baseRevision: NonNegativeIntegerSchema,
|
||||
messageId: BoundedMessageIdSchema,
|
||||
}),
|
||||
Schema.Struct({ type: Schema.Literal("session.end") }),
|
||||
Schema.Struct({
|
||||
type: Schema.Literal("ping"),
|
||||
nonce: BoundedMessageIdSchema,
|
||||
}),
|
||||
);
|
||||
|
||||
const ServerSyncMessageSchema = Schema.Union(
|
||||
Schema.Struct({
|
||||
type: Schema.Literal("location.snapshot"),
|
||||
snapshot: SnapshotSchema,
|
||||
originatingMessageId: Schema.NullOr(BoundedMessageIdSchema),
|
||||
}),
|
||||
Schema.Struct({
|
||||
type: Schema.Literal("presence.snapshot"),
|
||||
presence: PresenceSchema,
|
||||
}),
|
||||
Schema.Struct({
|
||||
type: Schema.Literal("location.rejected"),
|
||||
reason: Schema.Literal("stale_revision"),
|
||||
current: SnapshotSchema,
|
||||
messageId: BoundedMessageIdSchema,
|
||||
}),
|
||||
Schema.Struct({
|
||||
type: Schema.Literal("session.ended"),
|
||||
reason: Schema.Literal("presenter_ended", "expired"),
|
||||
}),
|
||||
Schema.Struct({
|
||||
type: Schema.Literal("protocol.error"),
|
||||
code: Schema.Literal("invalid_message", "message_too_large", "forbidden"),
|
||||
message: Schema.String,
|
||||
}),
|
||||
);
|
||||
|
||||
const CreateSessionRequestSchema = Schema.Struct({
|
||||
role: PresentationRoleSchema,
|
||||
initialHash: CanonicalHashSchema,
|
||||
});
|
||||
|
||||
const JoinSessionRequestSchema = Schema.Struct({
|
||||
role: PresentationRoleSchema,
|
||||
code: Schema.String.pipe(
|
||||
Schema.filter((value) => value.length === JOIN_CODE_LENGTH),
|
||||
),
|
||||
});
|
||||
|
||||
const parseJson = (input: string): DecodeResult<unknown> => {
|
||||
// Measure UTF-8 bytes before parsing so websocket limits match transport size.
|
||||
if (new TextEncoder().encode(input).byteLength > MAX_SYNC_MESSAGE_BYTES) {
|
||||
return { ok: false, error: "message_too_large" };
|
||||
}
|
||||
|
||||
try {
|
||||
return { ok: true, value: JSON.parse(input) as unknown };
|
||||
} catch {
|
||||
return { ok: false, error: "invalid_json" };
|
||||
}
|
||||
};
|
||||
|
||||
const decodeSchema = <T>(
|
||||
input: string,
|
||||
schema: Schema.Schema<T>,
|
||||
): DecodeResult<T> => {
|
||||
const parsed = parseJson(input);
|
||||
if (!parsed.ok) return parsed;
|
||||
|
||||
try {
|
||||
return {
|
||||
ok: true,
|
||||
value: Schema.decodeUnknownSync(schema, { onExcessProperty: "error" })(
|
||||
parsed.value,
|
||||
),
|
||||
};
|
||||
} catch {
|
||||
return { ok: false, error: "invalid_message" };
|
||||
}
|
||||
};
|
||||
|
||||
export const decodeClientSyncMessage = (
|
||||
input: string,
|
||||
): DecodeResult<ClientSyncMessage> =>
|
||||
decodeSchema(input, ClientSyncMessageSchema);
|
||||
|
||||
export const decodeServerSyncMessage = (
|
||||
input: string,
|
||||
): DecodeResult<ServerSyncMessage> =>
|
||||
decodeSchema(input, ServerSyncMessageSchema);
|
||||
|
||||
export const decodeCreateSessionRequest = (
|
||||
input: string,
|
||||
): DecodeResult<CreateSessionRequest> =>
|
||||
decodeSchema(input, CreateSessionRequestSchema);
|
||||
|
||||
export const decodeJoinSessionRequest = (
|
||||
input: string,
|
||||
): DecodeResult<JoinSessionRequest> => {
|
||||
const parsed = parseJson(input);
|
||||
if (!parsed.ok) return parsed;
|
||||
|
||||
const normalized =
|
||||
// Normalize before schema decoding so the exact length applies to user input
|
||||
// after harmless separators have been removed.
|
||||
typeof parsed.value === "object" &&
|
||||
parsed.value !== null &&
|
||||
!Array.isArray(parsed.value) &&
|
||||
"code" in parsed.value &&
|
||||
typeof parsed.value.code === "string"
|
||||
? { ...parsed.value, code: normalizeJoinCode(parsed.value.code) }
|
||||
: parsed.value;
|
||||
|
||||
try {
|
||||
return {
|
||||
ok: true,
|
||||
value: Schema.decodeUnknownSync(JoinSessionRequestSchema, {
|
||||
onExcessProperty: "error",
|
||||
})(normalized),
|
||||
};
|
||||
} catch {
|
||||
return { ok: false, error: "invalid_message" };
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"extends": "../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"module": "NodeNext",
|
||||
"moduleResolution": "NodeNext",
|
||||
"composite": true,
|
||||
"declaration": true,
|
||||
"lib": ["ES2023"],
|
||||
"types": ["node"],
|
||||
"rootDir": "src",
|
||||
"outDir": "dist"
|
||||
},
|
||||
"include": ["src"],
|
||||
"exclude": ["src/**/*.test.ts"]
|
||||
}
|
||||
Reference in New Issue
Block a user