export * as Event from "./event" import { Schema } from "effect" import { optional } from "./schema" import { ascending } from "./identifier" import { Location } from "./location" import { statics } from "./schema" export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe( Schema.brand("Event.ID"), statics((schema) => ({ create: () => schema.make("evt_" + ascending()) })), ) export type ID = typeof ID.Type export type Definition< Type extends string = string, DataSchema extends Schema.Codec = Schema.Codec, > = Schema.Top & { readonly type: Type readonly durable?: { readonly version: number readonly aggregate: string } readonly data: DataSchema } export type Data = Schema.Schema.Type export type Payload = { readonly id: ID readonly type: D["type"] readonly data: Data readonly durable?: { readonly aggregateID: string readonly seq: number readonly version: number } readonly location?: Location.Ref readonly metadata?: Record } export function define< const Type extends string, const Fields extends Readonly>>, >(input: { readonly type: Type readonly durable?: { readonly version: number readonly aggregate: string } readonly schema: Fields }) { const data = Schema.Struct(input.schema) return Schema.Struct({ id: ID, metadata: optional(Schema.Record(Schema.String, Schema.Unknown)), type: Schema.Literal(input.type), durable: optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Int, version: Schema.Int })), location: optional(Location.Ref), data, }) .annotate({ identifier: input.type }) .pipe( statics(() => ({ type: input.type, ...(input.durable === undefined ? {} : { durable: input.durable }), data, })), ) satisfies Definition } export function inventory>(...definitions: Definitions) { return Object.freeze(definitions) } export function latest(definitions: ReadonlyArray) { return readonlyMap( definitions.reduce((result, definition) => { const existing = result.get(definition.type) if (!existing) { result.set(definition.type, definition) return result } if (definition.durable && existing.durable && definition.durable.version !== existing.durable.version) { if (definition.durable.version > existing.durable.version) result.set(definition.type, definition) return result } if (definition !== existing) throw new Error(`Duplicate latest event definition for ${definition.type}`) return result }, new Map()), ) } export function versionedType(type: string, version: number) { return `${type}.${version}` } export function durable>(definitions: Definitions) { return readonlyMap( definitions.reduce((result, definition) => { if (!definition.durable) return result const key = versionedType(definition.type, definition.durable.version) if (result.has(key)) throw new Error(`Duplicate durable event definition for ${key}`) result.set(key, definition) return result }, new Map()), ) } function readonlyMap(map: Map): ReadonlyMap { const result: ReadonlyMap = Object.freeze({ get size() { return map.size }, entries: () => map.entries(), forEach: (callback: (value: Value, key: Key, map: ReadonlyMap) => void, thisArg?: unknown) => map.forEach((value, key) => callback.call(thisArg, value, key, result)), get: (key: Key) => map.get(key), has: (key: Key) => map.has(key), keys: () => map.keys(), values: () => map.values(), [Symbol.iterator]: () => map[Symbol.iterator](), }) return result }