26078594527

Committed 19 May 2026 05:40AM UTC coverage: 69.413% (-5.6%) from 74.971%

Build # 26078594527

Build Type

push

github

Committed by

cr15p1

Commit Message

Remove remaining duplicated Sonar blocks

Coverage Stats

1990 of 3317 branches covered (59.99%)

Branch coverage included in aggregate %.

9 of 23 new or added lines in 3 files covered. (39.13%)

304 existing lines in 15 files now uncovered.

4321 of 5775 relevant lines covered (74.82%)

25.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

62.35

/packages/schema/src/schemaModule.ts

import { createSchemaContext } from './context.js';
import {
  runFieldOperation,
  runOperation,
  type FieldOperation,
  type Operation,
  type OperationExecutor,
  type OperationPublishMap,
  type OperationRooms,
} from './operation.js';
import type {
  AckConfig,
  AstNode,
  Logger,
  PublishInput,
  PublishAck,
  SchemaContext,
  SchemaRequestContextInput,
  SchemaLike,
  Shape,
} from './types.js';
import { pack, unpack } from 'msgpackr';
import type { Subscription } from './api.js';

import type {
  EventEnvelope,
  EventError,
  RuntimeContext,
  RuntimeModule,
  RuntimeModuleRegister,
  RuntimeRegistry,
  PartialEventEnvelope,
} from '@livon/runtime';

type RuntimeNext = (update?: PartialEventEnvelope) => Promise<EventEnvelope>;

type AnySchema = SchemaLike;
type AnyInput = AnySchema | Shape | undefined;

type AnyOperation = Omit<Operation<AnySchema, AnySchema | undefined, unknown>, 'exec' | 'publish' | 'rooms'> & {
  exec: OperationExecutor<never, unknown>;
  publish?: OperationPublishMap<never>;
  rooms?: OperationRooms<never>;
};
type AnyFieldOperation = Omit<FieldOperation<AnySchema, AnyInput, AnySchema | undefined, unknown>, 'exec'> & {
  exec: unknown;
};
type AnySubscription = Subscription<AnySchema | undefined, AnySchema, AnySchema | undefined, unknown>;

export interface SchemaModuleLike {
  operations: Record<string, AnyOperation>;
  fieldOperations: Record<string, AnyFieldOperation>;
  subscriptions: Record<string, AnySubscription>;
  ast: () => AstNode;
}

export type SchemaModuleInput = SchemaModuleLike;

const normalizeSchemaModuleInput = (input: SchemaModuleLike): SchemaModuleInput => ({
  operations: input.operations,
  fieldOperations: input.fieldOperations,
  subscriptions: input.subscriptions,
  ast: input.ast,
});

export type SchemaModuleOptions = {
  explain?: boolean;
  schemaVersion?: string;
  now?: SchemaModuleNow;
  encoder?: SchemaModuleEncoder;
  decoder?: SchemaModuleDecoder;
  logger?: Logger;
  getRequestContext?: SchemaModuleGetRequestContext;
};

export interface SchemaModuleNow {
  (): number;
}

export interface SchemaModuleEncoder {
  (value: unknown): Uint8Array;
}

export interface SchemaModuleDecoder {
  (payload: Uint8Array | unknown): unknown;
}

export interface SchemaModuleGetRequestContext {
  (envelope: EventEnvelope, ctx: RuntimeContext): SchemaRequestContextInput | undefined;
}

export interface ExplainPayload {
  ast: AstNode;
  checksum: string;
  schemaVersion?: string;
  generatedAt: string;
  etag: string;
  notModified?: boolean;
}

export interface FieldPayload {
  dependsOn: unknown;
  input?: unknown;
}

export interface BuildExplainPayloadInput {
  notModified?: boolean;
  ast: AstNode;
  checksum: string;
  schemaVersion?: string;
  now: () => number;
}

interface HandleExplainRequestInput {
  ast: AstNode;
  checksum: string;
  ctx: RuntimeContext;
  encode: SchemaModuleEncoder;
  envelope: EventEnvelope;
  next: RuntimeNext;
  now: SchemaModuleNow;
  options: SchemaModuleOptions;
}

interface HandleExplainRequestResult {
  envelope?: EventEnvelope;
  handled: boolean;
}

interface ResolveFieldOperationInput {
  event: string;
  moduleSchema: SchemaModuleInput;
  operation?: AnyOperation;
}

type RuntimeOperation = Operation<AnySchema, AnySchema | undefined, unknown>;
type RuntimeFieldOperation = FieldOperation<AnySchema, AnyInput, AnySchema | undefined, unknown>;

export interface EmitErrorEventInput {
  ctx: RuntimeContext;
  envelope: EventEnvelope;
  metadata?: Readonly<Record<string, unknown>>;
  error: unknown;
  info?: Readonly<Record<string, unknown>>;
}

export type EnvelopeWithMeta = EventEnvelope & {
  meta?: Readonly<Record<string, unknown>>;
};

const isRecord = (value: unknown): value is Record<string, unknown> =>
  typeof value === 'object' && value !== null && !Array.isArray(value);

const defaultEncode = (value: unknown): Uint8Array => pack(value);

const defaultDecode = (payload: Uint8Array | unknown): unknown =>
  payload instanceof Uint8Array ? unpack(payload) : payload;

const stableStringify = (value: unknown): string => {
  if (value === null || typeof value !== 'object') {
    return JSON.stringify(value);
  }
  if (Array.isArray(value)) {
    return `[${value.map((entry) => stableStringify(entry)).join(',')}]`;
  }
  const record = value as Record<string, unknown>;
  const keys = Object.keys(record).sort((left, right) => left.localeCompare(right));
  return `{${keys.map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`).join(',')}}`;
};

const hashString = (input: string): string =>
  Array.from(input).reduce((hash, char) => {
    const next = (hash ^ char.charCodeAt(0)) >>> 0;
    return Math.imul(next, 16777619) >>> 0;
  }, 2166136261).toString(16).padStart(8, '0');

const getEnvelopeMetadata = (envelope: EventEnvelope): Readonly<Record<string, unknown>> | undefined => {
  if (envelope.metadata) {
    return envelope.metadata;
  }
  return (envelope as EnvelopeWithMeta).meta;
};

const normalizeFieldPayload = (payload: unknown): FieldPayload =>
  isRecord(payload) && 'dependsOn' in payload
    ? (payload as unknown as FieldPayload)
    : { dependsOn: payload };

interface FieldEventInfo {
  owner: string;
  field: string;
}

const splitFieldEvent = (event: string): FieldEventInfo | undefined => {
  if (!event.startsWith('$')) {
    return undefined;
  }
  const body = event.slice(1);
  const dotIndex = body.indexOf('.');
  if (dotIndex <= 0 || dotIndex === body.length - 1) {
    return undefined;
  }
  return { owner: body.slice(0, dotIndex), field: body.slice(dotIndex + 1) };
};

const mergeMetadata = (
  base?: Readonly<Record<string, unknown>>,
  extra?: Readonly<Record<string, unknown>>,
) => {
  if (!base && !extra) {
    return undefined;
  }
  return { ...(base ?? {}), ...(extra ?? {}) };
};

const buildExplainPayload = (input: BuildExplainPayloadInput): ExplainPayload => ({
  ast: input.ast,
  checksum: input.checksum,
  schemaVersion: input.schemaVersion,
  generatedAt: new Date(input.now()).toISOString(),
  etag: input.checksum,
  ...(input.notModified ? { notModified: true } : {}),
});

const handleExplainRequest = async ({
  ast,
  checksum,
  ctx,
  encode,
  envelope,
  next,
  now,
  options,
}: HandleExplainRequestInput): Promise<HandleExplainRequestResult> => {
  if (envelope.event !== '$explain') {
    return { handled: false };
  }

  if (!options.explain) {
    return { handled: true, envelope: await next() };
  }

  const metadata = getEnvelopeMetadata(envelope);
  const ifNoneMatch = typeof metadata?.ifNoneMatch === 'string' ? metadata.ifNoneMatch : undefined;
  const notModified = Boolean(ifNoneMatch && ifNoneMatch === checksum);
  const payload = buildExplainPayload({
    notModified,
    ast,
    checksum,
    schemaVersion: options.schemaVersion,
    now,
  });
  await ctx.emitEvent({
    event: envelope.event,
    payload: encode(payload),
    metadata,
    context: envelope.context ? { ...envelope.context } : undefined,
  });

  return { handled: true, envelope };
};

const resolveFieldOperation = ({
  event,
  moduleSchema,
  operation,
}: ResolveFieldOperationInput): AnyFieldOperation | undefined => {
  if (operation) {
    return undefined;
  }

  const fieldInfo = splitFieldEvent(event);
  if (!fieldInfo) {
    return undefined;
  }

  const fieldKey = `${fieldInfo.owner}.${fieldInfo.field}`;
  return moduleSchema.fieldOperations[fieldKey] ?? moduleSchema.fieldOperations[fieldInfo.field];
};

const runAnyOperation = (
  operation: AnyOperation,
  input: unknown,
  context: SchemaContext,
): Promise<unknown> =>
  runOperation(operation as unknown as RuntimeOperation, input, context);

const runAnyFieldOperation = (
  operation: AnyFieldOperation,
  dependsOn: unknown,
  input: unknown,
  context: SchemaContext,
): Promise<unknown> =>
  runFieldOperation(operation as unknown as RuntimeFieldOperation, dependsOn, input, context);

const eventErrorFromUnknown = (error: unknown, info?: Readonly<Record<string, unknown>>): EventError => {
  if (error instanceof Error) {
    return {
      message: error.message,
      name: error.name,
      stack: error.stack,
      context: info,
    };
  }
  return {
    message: typeof error === 'string' ? error : 'Unknown error',
    context: info,
  };
};

const normalizeAckConfig = (ack?: PublishAck): AckConfig | undefined => {
  if (ack === undefined || ack === false) {
    return undefined;
  }
  if (ack === true) {
    return { required: true, mode: 'received' };
  }
  return {
    required: ack.required ?? true,
    mode: ack.mode ?? 'received',
    ...(typeof ack.timeoutMs === 'number' ? { timeoutMs: ack.timeoutMs } : {}),
    ...(typeof ack.retries === 'number' ? { retries: ack.retries } : {}),
  };
};

interface ResolveSubscriptionPayloadInput {
  subscription: AnySubscription;
  input?: unknown;
  payload: unknown;
  ctx: SchemaContext;
}

interface ResolveSubscriptionPayloadResult {
  skip: boolean;
  payload?: unknown;
}

const resolveSubscriptionPayload = async (
  input: ResolveSubscriptionPayloadInput,
): Promise<ResolveSubscriptionPayloadResult> => {
  const parsedInput = input.subscription.input
    ? input.subscription.input.parse(input.input, input.ctx)
    : input.input;
  const parsedPayload = input.subscription.payload.parse(input.payload, input.ctx);
  if (input.subscription.filter) {
    const allowed = await input.subscription.filter(parsedInput, parsedPayload, input.ctx);
    if (!allowed) {
      return { skip: true };
    }
  }
  const executed = input.subscription.exec
    ? await input.subscription.exec(parsedInput, parsedPayload, input.ctx)
    : parsedPayload;
  const output = input.subscription.output
    ? input.subscription.output.parse(executed, input.ctx)
    : executed;
  return { skip: false, payload: output };
};

const emitErrorEvent = async (input: EmitErrorEventInput) =>
  input.ctx.emitError({
    id: input.envelope.id,
    event: input.envelope.event,
    status: input.envelope.status,
    ...(input.envelope.payload !== undefined ? { payload: input.envelope.payload } : {}),
    error: eventErrorFromUnknown(input.error, input.info),
    metadata: input.metadata,
    context: input.envelope.context ? { ...input.envelope.context } : undefined,
  });

/**
 * schemaModule is part of the public LIVON API.
 *
 * @remarks
 * Parameter and return types are defined in the TypeScript signature.
 *
 * @see https://livon.tech/docs/packages/schema
 *
 * @example
 * const result = schemaModule(undefined as never);
 */
export const schemaModule = (schema: SchemaModuleLike, options: SchemaModuleOptions = {}): RuntimeModule => {
  const moduleSchema = normalizeSchemaModuleInput(schema);
  const encode = options.encoder ?? defaultEncode;
  const decode = options.decoder ?? defaultDecode;
  const now = options.now ?? Date.now;

  const ast = moduleSchema.ast();
  const checksum = hashString(stableStringify(ast));

  const onReceive = async (envelope: EventEnvelope, ctx: RuntimeContext, next: RuntimeNext) => {
    const explainResult = await handleExplainRequest({
      ast,
      checksum,
      ctx,
      encode,
      envelope,
      next,
      now,
      options,
    });
    if (explainResult.handled) {
      return explainResult.envelope ?? envelope;
    }

    const op = moduleSchema.operations[envelope.event];
    const fieldOp = resolveFieldOperation({
      event: envelope.event,
      moduleSchema,
      operation: op,
    });

    if (!op && !fieldOp) {
      return next();
    }

    const metadata = getEnvelopeMetadata(envelope);
    const requestOverrides = options.getRequestContext ? options.getRequestContext(envelope, ctx) : undefined;
    const externalPublisher = requestOverrides?.publisher;
    const externalOnPublishError = requestOverrides?.onPublishError;
    const schemaContext = createSchemaContext();

    const publisher = async (input: PublishInput) => {
      const subscription = moduleSchema.subscriptions[input.topic];
      if (!subscription) {
        throw new Error(`schemaModule: publish topic "${input.topic}" has no matching subscription.`);
      }
      const resolved = await resolveSubscriptionPayload({
        subscription,
        input: input.input,
        payload: input.payload,
        ctx: schemaContext,
      });
      if (resolved.skip) {
        return;
      }
      const keyMeta = input.key ? { key: input.key } : undefined;
      const ackMeta = normalizeAckConfig(input.ack);
      const ackWrapper = ackMeta ? { ack: ackMeta } : undefined;
      const publishMeta = mergeMetadata(
        metadata,
        mergeMetadata(
          input.meta,
          mergeMetadata(keyMeta, ackWrapper),
        ),
      );
      await ctx.emitEvent({
        event: input.topic,
        payload: encode(resolved.payload),
        metadata: publishMeta,
        context: envelope.context ? { ...envelope.context } : undefined,
      });
      if (externalPublisher) {
        await externalPublisher({
          ...input,
          payload: resolved.payload,
          ...(ackMeta ? { ack: ackMeta } : {}),
        });
      }
    };

    const handlePublishError = (error: unknown, info?: Readonly<Record<string, unknown>>) => {
      if (externalOnPublishError) {
        externalOnPublishError(error, info);
      }
      void emitErrorEvent({ ctx, envelope, metadata, error, info });
    };

    const requestContext: SchemaRequestContextInput = {
      ...requestOverrides,
      metadata: mergeMetadata(metadata, requestOverrides?.metadata),
      publisher,
      onPublishError: handlePublishError,
      logger: requestOverrides?.logger ?? options.logger,
    };

    schemaContext.setRequestContext(requestContext);
    const input = decode(envelope.payload);
    const fieldPayload = op ? undefined : normalizeFieldPayload(input);
    let result: unknown;
    try {
      result = op
        ? await runAnyOperation(op, input, schemaContext)
        : await runAnyFieldOperation(
            fieldOp as AnyFieldOperation,
            fieldPayload?.dependsOn,
            fieldPayload?.input,
            schemaContext,
          );
    } catch (error) {
      await emitErrorEvent({
        ctx,
        envelope,
        metadata,
        error,
        info: { phase: 'execute', event: envelope.event },
      });
      return envelope;
    }

    await ctx.emitEvent({
      event: envelope.event,
      payload: encode(result),
      metadata,
      context: envelope.context ? { ...envelope.context } : undefined,
    });

    return envelope;
  };

  const register: RuntimeModuleRegister = (registry: RuntimeRegistry) => {
    registry.onReceive(onReceive);
  };

  return {
    name: 'schema',
    register,
  };
};

1	import { createSchemaContext } from './context.js';
2	import {
3	runFieldOperation,
4	runOperation,
5	type FieldOperation,
6	type Operation,
7	type OperationExecutor,
8	type OperationPublishMap,
9	type OperationRooms,
10	} from './operation.js';
11	import type {
12	AckConfig,
13	AstNode,
14	Logger,
15	PublishInput,
16	PublishAck,
17	SchemaContext,
18	SchemaRequestContextInput,
19	SchemaLike,
20	Shape,
21	} from './types.js';
22	import { pack, unpack } from 'msgpackr';
23	import type { Subscription } from './api.js';
24
25	import type {
26	EventEnvelope,
27	EventError,
28	RuntimeContext,
29	RuntimeModule,
30	RuntimeModuleRegister,
31	RuntimeRegistry,
32	PartialEventEnvelope,
33	} from '@livon/runtime';
34
35	type RuntimeNext = (update?: PartialEventEnvelope) => Promise<EventEnvelope>;
36
37	type AnySchema = SchemaLike;
38	type AnyInput = AnySchema \| Shape \| undefined;
39
40	type AnyOperation = Omit<Operation<AnySchema, AnySchema \| undefined, unknown>, 'exec' \| 'publish' \| 'rooms'> & {
41	exec: OperationExecutor<never, unknown>;
42	publish?: OperationPublishMap<never>;
43	rooms?: OperationRooms<never>;
44	};
45	type AnyFieldOperation = Omit<FieldOperation<AnySchema, AnyInput, AnySchema \| undefined, unknown>, 'exec'> & {
46	exec: unknown;
47	};
48	type AnySubscription = Subscription<AnySchema \| undefined, AnySchema, AnySchema \| undefined, unknown>;
49
50	export interface SchemaModuleLike {
51	operations: Record<string, AnyOperation>;
52	fieldOperations: Record<string, AnyFieldOperation>;
53	subscriptions: Record<string, AnySubscription>;
54	ast: () => AstNode;
55	}
56
57	export type SchemaModuleInput = SchemaModuleLike;
58
59	const normalizeSchemaModuleInput = (input: SchemaModuleLike): SchemaModuleInput => ({	7✔
60	operations: input.operations,
61	fieldOperations: input.fieldOperations,
62	subscriptions: input.subscriptions,
63	ast: input.ast,
64	});
65
66	export type SchemaModuleOptions = {
67	explain?: boolean;
68	schemaVersion?: string;
69	now?: SchemaModuleNow;
70	encoder?: SchemaModuleEncoder;
71	decoder?: SchemaModuleDecoder;
72	logger?: Logger;
73	getRequestContext?: SchemaModuleGetRequestContext;
74	};
75
76	export interface SchemaModuleNow {
77	(): number;
78	}
79
80	export interface SchemaModuleEncoder {
81	(value: unknown): Uint8Array;
82	}
83
84	export interface SchemaModuleDecoder {
85	(payload: Uint8Array \| unknown): unknown;
86	}
87
88	export interface SchemaModuleGetRequestContext {
89	(envelope: EventEnvelope, ctx: RuntimeContext): SchemaRequestContextInput \| undefined;
90	}
91
92	export interface ExplainPayload {
93	ast: AstNode;
94	checksum: string;
95	schemaVersion?: string;
96	generatedAt: string;
97	etag: string;
98	notModified?: boolean;
99	}
100
101	export interface FieldPayload {
102	dependsOn: unknown;
103	input?: unknown;
104	}
105
106	export interface BuildExplainPayloadInput {
107	notModified?: boolean;
108	ast: AstNode;
109	checksum: string;
110	schemaVersion?: string;
111	now: () => number;
112	}
113
114	interface HandleExplainRequestInput {
115	ast: AstNode;
116	checksum: string;
117	ctx: RuntimeContext;
118	encode: SchemaModuleEncoder;
119	envelope: EventEnvelope;
120	next: RuntimeNext;
121	now: SchemaModuleNow;
122	options: SchemaModuleOptions;
123	}
124
125	interface HandleExplainRequestResult {
126	envelope?: EventEnvelope;
127	handled: boolean;
128	}
129
130	interface ResolveFieldOperationInput {
131	event: string;
132	moduleSchema: SchemaModuleInput;
133	operation?: AnyOperation;
134	}
135
136	type RuntimeOperation = Operation<AnySchema, AnySchema \| undefined, unknown>;
137	type RuntimeFieldOperation = FieldOperation<AnySchema, AnyInput, AnySchema \| undefined, unknown>;
138
139	export interface EmitErrorEventInput {
140	ctx: RuntimeContext;
141	envelope: EventEnvelope;
142	metadata?: Readonly<Record<string, unknown>>;
143	error: unknown;
144	info?: Readonly<Record<string, unknown>>;
145	}
146
147	export type EnvelopeWithMeta = EventEnvelope & {
148	meta?: Readonly<Record<string, unknown>>;
149	};
150
151	const isRecord = (value: unknown): value is Record<string, unknown> =>	3✔
152	typeof value === 'object' && value !== null && !Array.isArray(value);	1✔
153
154	const defaultEncode = (value: unknown): Uint8Array => pack(value);	3✔
155
156	const defaultDecode = (payload: Uint8Array \| unknown): unknown =>	3✔
UNCOV 157	payload instanceof Uint8Array ? unpack(payload) : payload;	×
158
159	const stableStringify = (value: unknown): string => {	3✔
160	if (value === null \|\| typeof value !== 'object') {	30✔
161	return JSON.stringify(value);	18✔
162	}
163	if (Array.isArray(value)) {	12✔
164	return `[${value.map((entry) => stableStringify(entry)).join(',')}]`;	2✔
165	}
166	const record = value as Record<string, unknown>;	10✔
167	const keys = Object.keys(record).sort((left, right) => left.localeCompare(right));	18✔
168	return `{${keys.map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`).join(',')}}`;	21✔
169	};
170
171	const hashString = (input: string): string =>	3✔
172	Array.from(input).reduce((hash, char) => {	7✔
173	const next = (hash ^ char.charCodeAt(0)) >>> 0;	354✔
174	return Math.imul(next, 16777619) >>> 0;	354✔
175	}, 2166136261).toString(16).padStart(8, '0');
176
177	const getEnvelopeMetadata = (envelope: EventEnvelope): Readonly<Record<string, unknown>> \| undefined => {	3✔
178	if (envelope.metadata) {	4✔
179	return envelope.metadata;	1✔
180	}
181	return (envelope as EnvelopeWithMeta).meta;	3✔
182	};
183
184	const normalizeFieldPayload = (payload: unknown): FieldPayload =>	3✔
185	isRecord(payload) && 'dependsOn' in payload	1!
186	? (payload as unknown as FieldPayload)
187	: { dependsOn: payload };
188
189	interface FieldEventInfo {
190	owner: string;
191	field: string;
192	}
193
194	const splitFieldEvent = (event: string): FieldEventInfo \| undefined => {	3✔
195	if (!event.startsWith('$')) {	2✔
196	return undefined;	1✔
197	}
198	const body = event.slice(1);	1✔
199	const dotIndex = body.indexOf('.');	1✔
200	if (dotIndex <= 0 \|\| dotIndex === body.length - 1) {	1!
UNCOV 201	return undefined;	×
202	}
203	return { owner: body.slice(0, dotIndex), field: body.slice(dotIndex + 1) };	1✔
204	};
205
206	const mergeMetadata = (	3✔
207	base?: Readonly<Record<string, unknown>>,
208	extra?: Readonly<Record<string, unknown>>,
209	) => {
210	if (!base && !extra) {	3!
211	return undefined;	3✔
212	}
UNCOV 213	return { ...(base ?? {}), ...(extra ?? {}) };	×
214	};
215
216	const buildExplainPayload = (input: BuildExplainPayloadInput): ExplainPayload => ({	3✔
217	ast: input.ast,
218	checksum: input.checksum,
219	schemaVersion: input.schemaVersion,
220	generatedAt: new Date(input.now()).toISOString(),
221	etag: input.checksum,
222	...(input.notModified ? { notModified: true } : {}),	1!
223	});
224
225	const handleExplainRequest = async ({	3✔
226	ast,
227	checksum,
228	ctx,
229	encode,
230	envelope,
231	next,
232	now,
233	options,
234	}: HandleExplainRequestInput): Promise<HandleExplainRequestResult> => {
235	if (envelope.event !== '$explain') {	5✔
236	return { handled: false };	4✔
237	}
238
239	if (!options.explain) {	1!
240	return { handled: true, envelope: await next() };	×
241	}
242
243	const metadata = getEnvelopeMetadata(envelope);	1✔
244	const ifNoneMatch = typeof metadata?.ifNoneMatch === 'string' ? metadata.ifNoneMatch : undefined;	1!
245	const notModified = Boolean(ifNoneMatch && ifNoneMatch === checksum);	5✔
246	const payload = buildExplainPayload({	5✔
247	notModified,
248	ast,
249	checksum,
250	schemaVersion: options.schemaVersion,
251	now,
252	});
253	await ctx.emitEvent({	5✔
254	event: envelope.event,
255	payload: encode(payload),
256	metadata,
257	context: envelope.context ? { ...envelope.context } : undefined,	1!
258	});
259
260	return { handled: true, envelope };	1✔
261	};
262
263	const resolveFieldOperation = ({	3✔
264	event,
265	moduleSchema,
266	operation,
267	}: ResolveFieldOperationInput): AnyFieldOperation \| undefined => {
268	if (operation) {	4✔
269	return undefined;	2✔
270	}
271
272	const fieldInfo = splitFieldEvent(event);	2✔
273	if (!fieldInfo) {	2✔
274	return undefined;	1✔
275	}
276
277	const fieldKey = `${fieldInfo.owner}.${fieldInfo.field}`;	1✔
278	return moduleSchema.fieldOperations[fieldKey] ?? moduleSchema.fieldOperations[fieldInfo.field];	1!
279	};
280
281	const runAnyOperation = (	3✔
282	operation: AnyOperation,
283	input: unknown,
284	context: SchemaContext,	2✔
285	): Promise<unknown> =>
286	runOperation(operation as unknown as RuntimeOperation, input, context);
287
288	const runAnyFieldOperation = (	3✔
289	operation: AnyFieldOperation,
290	dependsOn: unknown,
291	input: unknown,
292	context: SchemaContext,	1✔
293	): Promise<unknown> =>
294	runFieldOperation(operation as unknown as RuntimeFieldOperation, dependsOn, input, context);
295
296	const eventErrorFromUnknown = (error: unknown, info?: Readonly<Record<string, unknown>>): EventError => {	3✔
297	if (error instanceof Error) {	1!
298	return {	1✔
299	message: error.message,
300	name: error.name,
301	stack: error.stack,
302	context: info,
303	};
304	}
UNCOV 305	return {	×
306	message: typeof error === 'string' ? error : 'Unknown error',	×
307	context: info,
308	};
309	};
310
311	const normalizeAckConfig = (ack?: PublishAck): AckConfig \| undefined => {	3✔
UNCOV 312	if (ack === undefined \|\| ack === false) {	×
UNCOV 313	return undefined;	×
314	}
UNCOV 315	if (ack === true) {	×
UNCOV 316	return { required: true, mode: 'received' };	×
317	}
UNCOV 318	return {	×
319	required: ack.required ?? true,	×
320	mode: ack.mode ?? 'received',	×
321	...(typeof ack.timeoutMs === 'number' ? { timeoutMs: ack.timeoutMs } : {}),	×
322	...(typeof ack.retries === 'number' ? { retries: ack.retries } : {}),	×
323	};
324	};
325
326	interface ResolveSubscriptionPayloadInput {
327	subscription: AnySubscription;
328	input?: unknown;
329	payload: unknown;
330	ctx: SchemaContext;
331	}
332
333	interface ResolveSubscriptionPayloadResult {
334	skip: boolean;
335	payload?: unknown;
336	}
337
338	const resolveSubscriptionPayload = async (	3✔
339	input: ResolveSubscriptionPayloadInput,
340	): Promise<ResolveSubscriptionPayloadResult> => {
341	const parsedInput = input.subscription.input	×
342	? input.subscription.input.parse(input.input, input.ctx)
343	: input.input;
UNCOV 344	const parsedPayload = input.subscription.payload.parse(input.payload, input.ctx);	×
UNCOV 345	if (input.subscription.filter) {	×
UNCOV 346	const allowed = await input.subscription.filter(parsedInput, parsedPayload, input.ctx);	×
UNCOV 347	if (!allowed) {	×
UNCOV 348	return { skip: true };	×
349	}
350	}
UNCOV 351	const executed = input.subscription.exec	×
352	? await input.subscription.exec(parsedInput, parsedPayload, input.ctx)
353	: parsedPayload;
UNCOV 354	const output = input.subscription.output	×
355	? input.subscription.output.parse(executed, input.ctx)
356	: executed;
UNCOV 357	return { skip: false, payload: output };	×
358	};
359
360	const emitErrorEvent = async (input: EmitErrorEventInput) =>	3✔
361	input.ctx.emitError({	1✔
362	id: input.envelope.id,
363	event: input.envelope.event,
364	status: input.envelope.status,
365	...(input.envelope.payload !== undefined ? { payload: input.envelope.payload } : {}),	1!
366	error: eventErrorFromUnknown(input.error, input.info),
367	metadata: input.metadata,
368	context: input.envelope.context ? { ...input.envelope.context } : undefined,	1!
369	});
370
371	/**
372	* schemaModule is part of the public LIVON API.
373	*
374	* @remarks
375	* Parameter and return types are defined in the TypeScript signature.
376	*
377	* @see https://livon.tech/docs/packages/schema
378	*
379	* @example
380	* const result = schemaModule(undefined as never);
381	*/
382	export const schemaModule = (schema: SchemaModuleLike, options: SchemaModuleOptions = {}): RuntimeModule => {	3✔
383	const moduleSchema = normalizeSchemaModuleInput(schema);	7✔
384	const encode = options.encoder ?? defaultEncode;	7✔
385	const decode = options.decoder ?? defaultDecode;	7✔
386	const now = options.now ?? Date.now;	7✔
387
388	const ast = moduleSchema.ast();	7✔
389	const checksum = hashString(stableStringify(ast));	7✔
390
391	const onReceive = async (envelope: EventEnvelope, ctx: RuntimeContext, next: RuntimeNext) => {	7✔
392	const explainResult = await handleExplainRequest({	5✔
393	ast,
394	checksum,
395	ctx,
396	encode,
397	envelope,
398	next,
399	now,
400	options,
401	});
402	if (explainResult.handled) {	5✔
403	return explainResult.envelope ?? envelope;	1!
404	}
405
406	const op = moduleSchema.operations[envelope.event];	4✔
407	const fieldOp = resolveFieldOperation({	4✔
408	event: envelope.event,
409	moduleSchema,
410	operation: op,
411	});
412
413	if (!op && !fieldOp) {	4✔
414	return next();	1✔
415	}
416
417	const metadata = getEnvelopeMetadata(envelope);	3✔
418	const requestOverrides = options.getRequestContext ? options.getRequestContext(envelope, ctx) : undefined;	3!
419	const externalPublisher = requestOverrides?.publisher;	5✔
420	const externalOnPublishError = requestOverrides?.onPublishError;	5✔
421	const schemaContext = createSchemaContext();	5✔
422
423	const publisher = async (input: PublishInput) => {	5✔
UNCOV 424	const subscription = moduleSchema.subscriptions[input.topic];	×
UNCOV 425	if (!subscription) {	×
UNCOV 426	throw new Error(`schemaModule: publish topic "${input.topic}" has no matching subscription.`);	×
427	}
UNCOV 428	const resolved = await resolveSubscriptionPayload({	×
429	subscription,
430	input: input.input,
431	payload: input.payload,
432	ctx: schemaContext,
433	});
UNCOV 434	if (resolved.skip) {	×
UNCOV 435	return;	×
436	}
UNCOV 437	const keyMeta = input.key ? { key: input.key } : undefined;	×
UNCOV 438	const ackMeta = normalizeAckConfig(input.ack);	×
UNCOV 439	const ackWrapper = ackMeta ? { ack: ackMeta } : undefined;	×
UNCOV 440	const publishMeta = mergeMetadata(	×
441	metadata,
442	mergeMetadata(
443	input.meta,
444	mergeMetadata(keyMeta, ackWrapper),
445	),
446	);
UNCOV 447	await ctx.emitEvent({	×
448	event: input.topic,
449	payload: encode(resolved.payload),
450	metadata: publishMeta,
451	context: envelope.context ? { ...envelope.context } : undefined,	×
452	});
UNCOV 453	if (externalPublisher) {	×
UNCOV 454	await externalPublisher({	×
455	...input,
456	payload: resolved.payload,
457	...(ackMeta ? { ack: ackMeta } : {}),	×
458	});
459	}
460	};
461
462	const handlePublishError = (error: unknown, info?: Readonly<Record<string, unknown>>) => {	5✔
UNCOV 463	if (externalOnPublishError) {	×
UNCOV 464	externalOnPublishError(error, info);	×
465	}
UNCOV 466	void emitErrorEvent({ ctx, envelope, metadata, error, info });	×
467	};
468
469	const requestContext: SchemaRequestContextInput = {	5✔
470	...requestOverrides,
471	metadata: mergeMetadata(metadata, requestOverrides?.metadata),
472	publisher,
473	onPublishError: handlePublishError,
474	logger: requestOverrides?.logger ?? options.logger,	8✔
475	};
476
477	schemaContext.setRequestContext(requestContext);	5✔
478	const input = decode(envelope.payload);	5✔
479	const fieldPayload = op ? undefined : normalizeFieldPayload(input);	5✔
480	let result: unknown;
481	try {	5✔
482	result = op	5✔
483	? await runAnyOperation(op, input, schemaContext)
484	: await runAnyFieldOperation(
485	fieldOp as AnyFieldOperation,
486	fieldPayload?.dependsOn,
487	fieldPayload?.input,
488	schemaContext,
489	);
490	} catch (error) {
491	await emitErrorEvent({	1✔
492	ctx,
493	envelope,
494	metadata,
495	error,
496	info: { phase: 'execute', event: envelope.event },
497	});
498	return envelope;	1✔
499	}
500
501	await ctx.emitEvent({	2✔
502	event: envelope.event,
503	payload: encode(result),
504	metadata,
505	context: envelope.context ? { ...envelope.context } : undefined,	2!
506	});
507
508	return envelope;	2✔
509	};
510
511	const register: RuntimeModuleRegister = (registry: RuntimeRegistry) => {	7✔
512	registry.onReceive(onReceive);	7✔
513	};
514
515	return {	7✔
516	name: 'schema',
517	register,
518	};
519	};

live-input-vector-output-node / livon-ts / 26078594527

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous