Publisher backends

Hidden page. It waits for review: the sidebar, search and search engines leave it out, and listed pages don't link to it.

No backend packages ship; these recipes are yours to copy and adapt (examples/realtime runs both, across servers). Both serialize events with danSON (so Dates survive) and share one helper that replays kept events after lastEventId, then follows live ones:

import { import EffectEffect, import LayerLayer, import StreamStream } from 'effect';
import { function createSerializer(opts?: SerializerOptions): SerializercreateSerializer } from '@trpc12/trpc/serializer';
import {
  function tracked<TData>(id: string, data: TData): TrackedEnvelope<TData>tracked,
  type type PublisherEvents = {
    [x: string]: unknown;
}PublisherEvents,
  type interface PublisherShape<TEvents extends PublisherEvents>PublisherShape,
} from '@trpc12/trpc/server';

const const danson: Serializer<DansonForm>danson = function createSerializer(opts?: SerializerOptions): SerializercreateSerializer();
const const encode: (data: unknown) => stringencode = (data: unknowndata: unknown) => var JSON: JSONJSON.JSON.stringify(value: any, replacer?: (this: any, key: string, value: any) => any, space?: string | number): string (+1 overload)stringify(const danson: Serializer<DansonForm>danson.Serializer<DansonForm>.serialize(value: unknown, opts?: SerializeOpts): SerializeReturnserialize(data: unknowndata));
const const decode: (text: string) => unknowndecode = (text: stringtext: string) => const danson: Serializer<DansonForm>danson.Serializer<DansonForm>.deserialize<unknown>(value: SerializeReturn, opts?: DeserializeOpts): unknowndeserialize(var JSON: JSONJSON.JSON.parse(text: string, reviver?: (this: any, key: string, value: any) => any): anyparse(text: stringtext));

const const resume: <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}) => Stream.Stream<TrackedEnvelope<unknown>, never, never>resume = <function (type parameter) TId in <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}): Stream.Stream<...>TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}opts: {
  live: Stream.Stream<[TId, string], never, never>live: import StreamStream.interface Stream<out A, out E = never, out R = never>Stream<[function (type parameter) TId in <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}): Stream.Stream<...>TId, string]>;
  replay: Effect.Effect<[TId, string][], never, never>replay: import EffectEffect.interface Effect<out A, out E = never, out R = never>Effect<[function (type parameter) TId in <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}): Stream.Stream<...>TId, string][]>;
  after: TId | undefinedafter: function (type parameter) TId in <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}): Stream.Stream<...>TId | undefined;
  newer: (a: TId, b: TId) => booleannewer: (a: TIda: function (type parameter) TId in <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}): Stream.Stream<...>TId, b: TIdb: function (type parameter) TId in <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}): Stream.Stream<...>TId) => boolean;
  key: (id: TId) => stringkey: (id: TIdid: function (type parameter) TId in <TId>(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}): Stream.Stream<...>TId) => string;
}) =>
  import StreamStream.const unwrap: <TrackedEnvelope<unknown>, never, never, never, never>(effect: Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, never, never>) => Stream.Stream<TrackedEnvelope<unknown>, never, never>unwrap(
    import EffectEffect.const gen: <Effect.Effect<[TId, string][], never, never>, Stream.Stream<TrackedEnvelope<unknown>, never, never>>(f: () => Generator<Effect.Effect<[TId, string][], never, never>, Stream.Stream<TrackedEnvelope<unknown>, never, never>, never>) => Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, never, never> (+1 overload)gen(function* () {
      const const replay: [TId, string][]replay = opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}opts.after: TId | undefinedafter === var undefinedundefined ? [] : yield* opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}opts.replay: Effect.Effect<[TId, string][], never, never>replay;
      const const seen: TId | undefinedseen = const replay: [TId, string][]replay.Array<[TId, string]>.at(index: number): [TId, string] | undefinedat(-1)?.[0] ?? opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}opts.after: TId | undefinedafter;
      const const live: Stream.Stream<[TId, string], never, never>live = opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}opts.live: Stream.Stream<[TId, string], never, never>live.Pipeable.pipe<Stream.Stream<[TId, string], never, never>, Stream.Stream<[TId, string], never, never>>(this: Stream.Stream<[TId, string], never, never>, ab: (_: Stream.Stream<[TId, string], never, never>) => Stream.Stream<[TId, string], never, never>): Stream.Stream<[TId, string], never, never> (+21 overloads)pipe(
        import StreamStream.const filter: <[TId, string]>(predicate: Predicate<[TId, string]>) => <E, R>(self: Stream.Stream<[TId, string], E, R>) => Stream.Stream<[TId, string], E, R> (+3 overloads)filter(([id: TIdid]) => const seen: TId | undefinedseen === var undefinedundefined || opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}opts.newer: (a: TId, b: TId) => booleannewer(id: TIdid, const seen: TId & ({} | null)seen)),
      );
      return import StreamStream.const concat: <[TId, string], never, never, [TId, string], never, never>(self: Stream.Stream<[TId, string], never, never>, that: Stream.Stream<[TId, string], never, never>) => Stream.Stream<[TId, string], never, never> (+1 overload)concat(import StreamStream.const fromIterable: <[TId, string]>(iterable: Iterable<[TId, string]>, options?: {
    readonly chunkSize?: number | undefined;
}) => Stream.Stream<[TId, string], never, never>fromIterable(const replay: [TId, string][]replay), const live: Stream.Stream<[TId, string], never, never>live).Pipeable.pipe<Stream.Stream<[TId, string], never, never>, Stream.Stream<TrackedEnvelope<unknown>, never, never>>(this: Stream.Stream<[TId, string], never, never>, ab: (_: Stream.Stream<[TId, string], never, never>) => Stream.Stream<TrackedEnvelope<unknown>, never, never>): Stream.Stream<TrackedEnvelope<unknown>, never, never> (+21 overloads)pipe(
        import StreamStream.const map: <[TId, string], TrackedEnvelope<unknown>>(f: (a: [TId, string], i: number) => TrackedEnvelope<unknown>) => <E, R>(self: Stream.Stream<[TId, string], E, R>) => Stream.Stream<TrackedEnvelope<unknown>, E, R> (+1 overload)map(([id: TIdid, payload: stringpayload]) => tracked<unknown>(id: string, data: unknown): TrackedEnvelope<unknown>tracked(opts: {
    live: Stream.Stream<[TId, string]>;
    replay: Effect.Effect<[TId, string][]>;
    after: TId | undefined;
    newer: (a: TId, b: TId) => boolean;
    key: (id: TId) => string;
}opts.key: (id: TId) => stringkey(id: TIdid), const decode: (text: string) => unknowndecode(payload: stringpayload))),
      );
    }),
  );

Redis

Effect's Redis service (effect/persistence) with NodeRedis.layer from @effect/platform-node. Each channel is a Redis stream, which keeps history and assigns ids, plus a pub/sub channel for live delivery. Subscribing before reading history means no event falls in between.

import { import NodeRedisNodeRedis } from '@effect/platform-node';
import { import RedisRedis } from 'effect/persistence';

const const redisNewer: (a: string, b: string) => booleanredisNewer = (a: stringa: string, b: stringb: string) => {
  const [const am: numberam = 0, const as: numberas = 0] = a: stringa.String.split(separator: string | RegExp, limit?: number): string[] (+1 overload)split('-').Array<string>.map<number>(callbackfn: (value: string, index: number, array: string[]) => number, thisArg?: any): number[]map(var Number: NumberConstructorNumber);
  const [const bm: numberbm = 0, const bs: numberbs = 0] = b: stringb.String.split(separator: string | RegExp, limit?: number): string[] (+1 overload)split('-').Array<string>.map<number>(callbackfn: (value: string, index: number, array: string[]) => number, thisArg?: any): number[]map(var Number: NumberConstructorNumber);
  return const am: numberam !== const bm: numberbm ? const am: numberam > const bm: numberbm : const as: numberas > const bs: numberbs;
};

const const redisPublisher: <TEvents extends PublisherEvents>(opts: {
    prefix: string;
    maxLen?: number;
}) => Effect.Effect<PublisherShape<TEvents>, never, Redis.Redis>redisPublisher = <function (type parameter) TEvents in <TEvents extends PublisherEvents>(opts: {
    prefix: string;
    maxLen?: number;
}): Effect.Effect<PublisherShape<TEvents>, never, Redis.Redis>TEvents extends type PublisherEvents = {
    [x: string]: unknown;
}PublisherEvents>(opts: {
    prefix: string;
    maxLen?: number;
}opts: {
  prefix: stringprefix: string;
  maxLen?: number | undefinedmaxLen?: number;
}) =>
  import EffectEffect.const gen: <Effect.Effect<{
    readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, Redis.RedisError>;
    readonly subscribe: (channel: string) => Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>;
    readonly eval: <Config extends {
        readonly params: ReadonlyArray<unknown>;
        readonly result: unknown;
    }>(script: Redis.Script<Config>) => (...params: Config["params"]) => Effect.Effect<Config["result"], Redis.RedisError>;
}, never, Redis.Redis>, PublisherShape<...>>(f: () => Generator<...>) => Effect.Effect<...> (+1 overload)gen(function* () {
    const const redis: {
    readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, Redis.RedisError>;
    readonly subscribe: (channel: string) => Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>;
    readonly eval: <Config extends {
        readonly params: ReadonlyArray<unknown>;
        readonly result: unknown;
    }>(script: Redis.Script<Config>) => (...params: Config["params"]) => Effect.Effect<Config["result"], Redis.RedisError>;
}redis = yield* import RedisRedis.class RedisRedis;
    const const key: (channel: string) => stringkey = (channel: stringchannel: string) => `${opts: {
    prefix: string;
    maxLen?: number;
}opts.prefix: stringprefix}:${channel: stringchannel}`;
    const const shape: {
    publish: (channel: string, data: unknown) => Effect.Effect<string, never, never>;
    subscribe: (channel: string, o?: {
        readonly lastEventId?: string | undefined;
    }) => Stream.Stream<TrackedEnvelope<unknown>, never, never>;
}shape = {
      publish: (channel: string, data: unknown) => Effect.Effect<string, never, never>publish: (channel: stringchannel: string, data: unknowndata: unknown) =>
        import EffectEffect.const gen: <Effect.Effect<unknown, Redis.RedisError, never>, string>(f: () => Generator<Effect.Effect<unknown, Redis.RedisError, never>, string, never>) => Effect.Effect<string, Redis.RedisError, never> (+1 overload)gen(function* () {
          const const payload: stringpayload = const encode: (data: unknown) => stringencode(data: unknowndata);
          const const id: stringid = yield* const redis: {
    readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, Redis.RedisError>;
    readonly subscribe: (channel: string) => Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>;
    readonly eval: <Config extends {
        readonly params: ReadonlyArray<unknown>;
        readonly result: unknown;
    }>(script: Redis.Script<Config>) => (...params: Config["params"]) => Effect.Effect<Config["result"], Redis.RedisError>;
}redis.send: <string>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<string, Redis.RedisError, never>send<string>(
            'XADD',
            const key: (channel: string) => stringkey(channel: stringchannel),
            'MAXLEN',
            '~',
            var String: StringConstructor
(value?: any) => stringString(opts: {
    prefix: string;
    maxLen?: number;
}opts.maxLen?: number | undefinedmaxLen ?? 1000),
            '*',
            'data',
            const payload: stringpayload,
          );
          yield* const redis: {
    readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, Redis.RedisError>;
    readonly subscribe: (channel: string) => Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>;
    readonly eval: <Config extends {
        readonly params: ReadonlyArray<unknown>;
        readonly result: unknown;
    }>(script: Redis.Script<Config>) => (...params: Config["params"]) => Effect.Effect<Config["result"], Redis.RedisError>;
}redis.send: <unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<unknown, Redis.RedisError, never>send(
            'PUBLISH',
            const key: (channel: string) => stringkey(channel: stringchannel),
            var JSON: JSONJSON.JSON.stringify(value: any, replacer?: (this: any, key: string, value: any) => any, space?: string | number): string (+1 overload)stringify([const id: stringid, const payload: stringpayload]),
          );
          return const id: stringid;
        }).Pipeable.pipe<Effect.Effect<string, Redis.RedisError, never>, Effect.Effect<string, never, never>>(this: Effect.Effect<string, Redis.RedisError, never>, ab: (_: Effect.Effect<string, Redis.RedisError, never>) => Effect.Effect<string, never, never>): Effect.Effect<string, never, never> (+21 overloads)pipe(import EffectEffect.const orDie: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, never, R>orDie),
      subscribe: (channel: string, o?: {
    readonly lastEventId?: string | undefined;
}) => Stream.Stream<TrackedEnvelope<unknown>, never, never>subscribe: (
        channel: stringchannel: string,
        o: {
    readonly lastEventId?: string | undefined;
} | undefinedo?: { readonly lastEventId?: string | undefinedlastEventId?: string | undefined },
      ) =>
        import StreamStream.const unwrap: <TrackedEnvelope<unknown>, never, never, never, Scope>(effect: Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, never, Scope>) => Stream.Stream<TrackedEnvelope<unknown>, never, never>unwrap(
          import EffectEffect.const gen: <Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>, Stream.Stream<TrackedEnvelope<unknown>, never, never>>(f: () => Generator<Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>, Stream.Stream<TrackedEnvelope<unknown>, never, never>, never>) => Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, never, Scope> (+1 overload)gen(function* () {
            const const messages: Dequeue<Redis.RedisMessage, Redis.RedisError>messages = yield* const redis: {
    readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, Redis.RedisError>;
    readonly subscribe: (channel: string) => Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>;
    readonly eval: <Config extends {
        readonly params: ReadonlyArray<unknown>;
        readonly result: unknown;
    }>(script: Redis.Script<Config>) => (...params: Config["params"]) => Effect.Effect<Config["result"], Redis.RedisError>;
}redis.subscribe: (channel: string) => Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>subscribe(const key: (channel: string) => stringkey(channel: stringchannel));
            return const resume: <string>(opts: {
    live: Stream.Stream<[string, string], never, never>;
    replay: Effect.Effect<[string, string][], never, never>;
    after: string | undefined;
    newer: (a: string, b: string) => boolean;
    key: (id: string) => string;
}) => Stream.Stream<TrackedEnvelope<unknown>, never, never>resume({
              live: Stream.Stream<[string, string], never, never>live: import StreamStream.const fromQueue: <Redis.RedisMessage, Redis.RedisError>(queue: Dequeue<Redis.RedisMessage, Redis.RedisError>) => Stream.Stream<Redis.RedisMessage, Redis.RedisError, never>fromQueue(const messages: Dequeue<Redis.RedisMessage, Redis.RedisError>messages).Pipeable.pipe<Stream.Stream<Redis.RedisMessage, Redis.RedisError, never>, Stream.Stream<[string, string], Redis.RedisError, never>, Stream.Stream<[string, string], never, never>>(this: Stream.Stream<...>, ab: (_: Stream.Stream<Redis.RedisMessage, Redis.RedisError, never>) => Stream.Stream<[string, string], Redis.RedisError, never>, bc: (_: Stream.Stream<[string, string], Redis.RedisError, never>) => Stream.Stream<[string, string], never, never>): Stream.Stream<...> (+21 overloads)pipe(
                import StreamStream.const map: <Redis.RedisMessage, [string, string]>(f: (a: Redis.RedisMessage, i: number) => [string, string]) => <E, R>(self: Stream.Stream<Redis.RedisMessage, E, R>) => Stream.Stream<[string, string], E, R> (+1 overload)map((m: Redis.RedisMessagem) => var JSON: JSONJSON.JSON.parse(text: string, reviver?: (this: any, key: string, value: any) => any): anyparse(m: Redis.RedisMessagem.RedisMessage.message: stringmessage) as [string, string]),
                import StreamStream.const orDie: <A, E, R>(self: Stream.Stream<A, E, R>) => Stream.Stream<A, never, R>orDie,
              ),
              replay: Effect.Effect<[string, string][], never, never>replay: const redis: {
    readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, Redis.RedisError>;
    readonly subscribe: (channel: string) => Effect.Effect<Dequeue<Redis.RedisMessage, Redis.RedisError>, never, Scope>;
    readonly eval: <Config extends {
        readonly params: ReadonlyArray<unknown>;
        readonly result: unknown;
    }>(script: Redis.Script<Config>) => (...params: Config["params"]) => Effect.Effect<Config["result"], Redis.RedisError>;
}redis
                .send: <[string, [string, string]][]>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<[string, [string, string]][], Redis.RedisError, never>send<[string, [string, string]][]>(
                  'XRANGE',
                  const key: (channel: string) => stringkey(channel: stringchannel),
                  `(${o: {
    readonly lastEventId?: string | undefined;
} | undefinedo?.lastEventId?: string | undefinedlastEventId}`,
                  '+',
                )
                .Pipeable.pipe<Effect.Effect<[string, [string, string]][], Redis.RedisError, never>, Effect.Effect<[string, string][], Redis.RedisError, never>, Effect.Effect<[string, string][], never, never>>(this: Effect.Effect<...>, ab: (_: Effect.Effect<[string, [string, string]][], Redis.RedisError, never>) => Effect.Effect<[string, string][], Redis.RedisError, never>, bc: (_: Effect.Effect<[string, string][], Redis.RedisError, never>) => Effect.Effect<[string, string][], never, never>): Effect.Effect<...> (+21 overloads)pipe(
                  import EffectEffect.const map: <[string, [string, string]][], [string, string][]>(f: (a: [string, [string, string]][]) => [string, string][]) => <E, R>(self: Effect.Effect<[string, [string, string]][], E, R>) => Effect.Effect<[string, string][], E, R> (+1 overload)map((entries: [string, [string, string]][]entries) =>
                    entries: [string, [string, string]][]entries.Array<[string, [string, string]]>.map<[string, string]>(callbackfn: (value: [string, [string, string]], index: number, array: [string, [string, string]][]) => [string, string], thisArg?: any): [string, string][]map(([id: stringid, [, payload: stringpayload]]): [string, string] => [
                      id: stringid,
                      payload: stringpayload,
                    ]),
                  ),
                  import EffectEffect.const orDie: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, never, R>orDie,
                ),
              after: string | undefinedafter: o: {
    readonly lastEventId?: string | undefined;
} | undefinedo?.lastEventId?: string | undefinedlastEventId,
              newer: (a: string, b: string) => booleannewer: const redisNewer: (a: string, b: string) => booleanredisNewer,
              key: (id: string) => stringkey: (id: stringid) => id: stringid,
            });
          }),
        ),
    };
    return const shape: {
    publish: (channel: string, data: unknown) => Effect.Effect<string, never, never>;
    subscribe: (channel: string, o?: {
        readonly lastEventId?: string | undefined;
    }) => Stream.Stream<TrackedEnvelope<unknown>, never, never>;
}shape as interface PublisherShape<TEvents extends PublisherEvents>PublisherShape<function (type parameter) TEvents in <TEvents extends PublisherEvents>(opts: {
    prefix: string;
    maxLen?: number;
}): Effect.Effect<PublisherShape<TEvents>, never, Redis.Redis>TEvents>;
  });

const const layer: Layer.Layer<Posts, Redis.RedisError, never>layer = import LayerLayer.const effect: <Posts, PublisherShape<PostEvents>>(service: Key<Posts, PublisherShape<PostEvents>>) => <E, R>(effect: Effect.Effect<PublisherShape<PostEvents>, E, R>) => Layer.Layer<Posts, E, Exclude<R, Scope>> (+1 overload)effect(class PostsPosts)(
  const redisPublisher: <PostEvents>(opts: {
    prefix: string;
    maxLen?: number;
}) => Effect.Effect<PublisherShape<PostEvents>, never, Redis.Redis>redisPublisher<type PostEvents = {
    created: Post;
    deleted: {
        id: string;
    };
}PostEvents>({ prefix: stringprefix: 'posts' }),
).Pipeable.pipe<Layer.Layer<Posts, never, Redis.Redis>, Layer.Layer<Posts, Redis.RedisError, never>>(this: Layer.Layer<Posts, never, Redis.Redis>, ab: (_: Layer.Layer<Posts, never, Redis.Redis>) => Layer.Layer<Posts, Redis.RedisError, never>): Layer.Layer<Posts, Redis.RedisError, never> (+21 overloads)pipe(import LayerLayer.const provide: <never, Redis.RedisError, Redis.Redis | NodeRedis.NodeRedis>(that: Layer.Layer<Redis.Redis | NodeRedis.NodeRedis, Redis.RedisError, never>) => <RIn2, E2, ROut2>(self: Layer.Layer<ROut2, E2, RIn2>) => Layer.Layer<ROut2, Redis.RedisError | E2, Exclude<RIn2, Redis.Redis | NodeRedis.NodeRedis>> (+3 overloads)provide(import NodeRedisNodeRedis.const layer: (options?: RedisClientOptions<RedisModules, RedisFunctions, RedisScripts, RespVersions, TypeMapping, RedisSocketOptions> | undefined) => Layer.Layer<Redis.Redis | NodeRedis.NodeRedis, Redis.RedisError>layer({ RedisClientOptions<M extends RedisModules = RedisModules, F extends RedisFunctions = RedisFunctions, S extends RedisScripts = RedisScripts, RESP extends RespVersions = 3, TYPE_MAPPING extends TypeMapping = TypeMapping, SocketOptions extends RedisSocketOptions = RedisSocketOptions>.url?: string | undefinedurl: var process: NodeJS.Processprocess.NodeJS.Process.env: NodeJS.ProcessEnvenv.string | undefinedREDIS_URL })));

Postgres

@effect/sql-pg's PgClient has listen and notify. Events go into a table for history; NOTIFY carries new ones. NOTIFY payloads are capped at 8000 bytes, so for bigger events send only the id and read the row.

import { import PgClientPgClient } from '@effect/sql-pg';
import { import RedactedRedacted } from 'effect';

// create table trpc_events (id bigserial primary key, channel text not null,
//   payload text not null, created_at timestamptz not null default now());

const const pgPublisher: <TEvents extends PublisherEvents>() => Effect.Effect<PublisherShape<TEvents>, never, PgClient.PgClient>pgPublisher = <function (type parameter) TEvents in <TEvents extends PublisherEvents>(): Effect.Effect<PublisherShape<TEvents>, never, PgClient.PgClient>TEvents extends type PublisherEvents = {
    [x: string]: unknown;
}PublisherEvents>() =>
  import EffectEffect.const gen: <Effect.Effect<PgClient.PgClient, never, PgClient.PgClient>, PublisherShape<TEvents>>(f: () => Generator<Effect.Effect<PgClient.PgClient, never, PgClient.PgClient>, PublisherShape<TEvents>, never>) => Effect.Effect<PublisherShape<TEvents>, never, PgClient.PgClient> (+1 overload)gen(function* () {
    const const sql: PgClient.PgClientsql = yield* import PgClientPgClient.const PgClient: Service<PgClient.PgClient, PgClient.PgClient>PgClient;
    const const shape: {
    publish: (channel: string, data: unknown) => Effect.Effect<string, never, never>;
    subscribe: (channel: string, o?: {
        readonly lastEventId?: string | undefined;
    }) => Stream.Stream<TrackedEnvelope<unknown>, never, never>;
}shape = {
      publish: (channel: string, data: unknown) => Effect.Effect<string, never, never>publish: (channel: stringchannel: string, data: unknowndata: unknown) =>
        import EffectEffect.const gen: <Effect.Effect<void, SqlError, never>, string>(f: () => Generator<Effect.Effect<void, SqlError, never>, string, never>) => Effect.Effect<string, SqlError, never> (+1 overload)gen(function* () {
          const const payload: stringpayload = const encode: (data: unknown) => stringencode(data: unknowndata);
          const [const row: {
    id: string;
}row] = yield* const sql: PgClient.PgClientsql<{ id: stringid: string }>`
            insert into trpc_events (channel, payload) values (${channel: stringchannel}, ${const payload: stringpayload})
            returning id::text`;
          yield* const sql: PgClient.PgClientsql.PgClient.notify: (channel: string, payload: string) => Effect.Effect<void, SqlError>notify(channel: stringchannel, var JSON: JSONJSON.JSON.stringify(value: any, replacer?: (this: any, key: string, value: any) => any, space?: string | number): string (+1 overload)stringify([const row: {
    id: string;
}row!.id: stringid, const payload: stringpayload]));
          return const row: {
    id: string;
}row!.id: stringid;
        }).Pipeable.pipe<Effect.Effect<string, SqlError, never>, Effect.Effect<string, never, never>>(this: Effect.Effect<string, SqlError, never>, ab: (_: Effect.Effect<string, SqlError, never>) => Effect.Effect<string, never, never>): Effect.Effect<string, never, never> (+21 overloads)pipe(import EffectEffect.const orDie: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, never, R>orDie),
      subscribe: (channel: string, o?: {
    readonly lastEventId?: string | undefined;
}) => Stream.Stream<TrackedEnvelope<unknown>, never, never>subscribe: (
        channel: stringchannel: string,
        o: {
    readonly lastEventId?: string | undefined;
} | undefinedo?: { readonly lastEventId?: string | undefinedlastEventId?: string | undefined },
      ) =>
        import StreamStream.const unwrap: <TrackedEnvelope<unknown>, never, never, never, Scope>(effect: Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, never, Scope>) => Stream.Stream<TrackedEnvelope<unknown>, never, never>unwrap(
          import EffectEffect.const gen: <Effect.Effect<Dequeue<Notification, SqlError>, SqlError, Scope>, Stream.Stream<TrackedEnvelope<unknown>, never, never>>(f: () => Generator<Effect.Effect<Dequeue<Notification, SqlError>, SqlError, Scope>, Stream.Stream<TrackedEnvelope<unknown>, never, never>, never>) => Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, SqlError, Scope> (+1 overload)gen(function* () {
            const const notifications: Dequeue<Notification, SqlError>notifications = yield* const sql: PgClient.PgClientsql.PgClient.listen: (channel: string) => Effect.Effect<Dequeue<Notification, SqlError>, SqlError, Scope>listen(channel: stringchannel);
            return const resume: <bigint>(opts: {
    live: Stream.Stream<[bigint, string], never, never>;
    replay: Effect.Effect<[bigint, string][], never, never>;
    after: bigint | undefined;
    newer: (a: bigint, b: bigint) => boolean;
    key: (id: bigint) => string;
}) => Stream.Stream<TrackedEnvelope<unknown>, never, never>resume({
              live: Stream.Stream<[bigint, string], never, never>live: import StreamStream.const fromQueue: <Notification, SqlError>(queue: Dequeue<Notification, SqlError>) => Stream.Stream<Notification, SqlError, never>fromQueue(const notifications: Dequeue<Notification, SqlError>notifications).Pipeable.pipe<Stream.Stream<Notification, SqlError, never>, Stream.Stream<[bigint, string], SqlError, never>, Stream.Stream<[bigint, string], never, never>>(this: Stream.Stream<Notification, SqlError, never>, ab: (_: Stream.Stream<Notification, SqlError, never>) => Stream.Stream<[bigint, string], SqlError, never>, bc: (_: Stream.Stream<[bigint, string], SqlError, never>) => Stream.Stream<[bigint, string], never, never>): Stream.Stream<...> (+21 overloads)pipe(
                import StreamStream.const map: <Notification, [bigint, string]>(f: (a: Notification, i: number) => [bigint, string]) => <E, R>(self: Stream.Stream<Notification, E, R>) => Stream.Stream<[bigint, string], E, R> (+1 overload)map((n: Notificationn) => {
                  const [const id: stringid, const payload: stringpayload] = var JSON: JSONJSON.JSON.parse(text: string, reviver?: (this: any, key: string, value: any) => any): anyparse(n: Notificationn.Notification.payload: stringpayload) as [
                    string,
                    string,
                  ];
                  return [var BigInt: BigIntConstructor
(value: bigint | boolean | number | string) => bigintBigInt(const id: stringid), const payload: stringpayload] as [bigint, string];
                }),
                import StreamStream.const orDie: <A, E, R>(self: Stream.Stream<A, E, R>) => Stream.Stream<A, never, R>orDie,
              ),
              replay: Effect.Effect<[bigint, string][], never, never>replay: const sql: PgClient.PgClientsql<{ id: stringid: string; payload: stringpayload: string }>`
                select id::text, payload from trpc_events
                where channel = ${channel: stringchannel} and id > ${o: {
    readonly lastEventId?: string | undefined;
} | undefinedo?.lastEventId?: string | undefinedlastEventId ?? '0'} order by id`.Pipeable.pipe<Statement<{
    id: string;
    payload: string;
}>, Effect.Effect<[bigint, string][], SqlError, never>, Effect.Effect<[bigint, string][], never, never>>(this: Statement<{
    id: string;
    payload: string;
}>, ab: (_: Statement<{
    id: string;
    payload: string;
}>) => Effect.Effect<[bigint, string][], SqlError, never>, bc: (_: Effect.Effect<[bigint, string][], SqlError, never>) => Effect.Effect<[bigint, string][], never, never>): Effect.Effect<[bigint, string][], never, never> (+21 overloads)pipe(
                import EffectEffect.const map: <readonly {
    id: string;
    payload: string;
}[], [bigint, string][]>(f: (a: readonly {
    id: string;
    payload: string;
}[]) => [bigint, string][]) => <E, R>(self: Effect.Effect<readonly {
    id: string;
    payload: string;
}[], E, R>) => Effect.Effect<[bigint, string][], E, R> (+1 overload)map((rows: readonly {
    id: string;
    payload: string;
}[]rows) =>
                  rows: readonly {
    id: string;
    payload: string;
}[]rows.ReadonlyArray<{ id: string; payload: string; }>.map<[bigint, string]>(callbackfn: (value: {
    id: string;
    payload: string;
}, index: number, array: readonly {
    id: string;
    payload: string;
}[]) => [bigint, string], thisArg?: any): [bigint, string][]map((r: {
    id: string;
    payload: string;
}r): [bigint, string] => [var BigInt: BigIntConstructor
(value: bigint | boolean | number | string) => bigintBigInt(r: {
    id: string;
    payload: string;
}r.id: stringid), r: {
    id: string;
    payload: string;
}r.payload: stringpayload]),
                ),
                import EffectEffect.const orDie: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, never, R>orDie,
              ),
              after: bigint | undefinedafter:
                o: {
    readonly lastEventId?: string | undefined;
} | undefinedo?.lastEventId?: string | undefinedlastEventId === var undefinedundefined
                  ? var undefinedundefined
                  : var BigInt: BigIntConstructor
(value: bigint | boolean | number | string) => bigintBigInt(o: {
    readonly lastEventId?: string | undefined;
}o.lastEventId?: stringlastEventId),
              newer: (a: bigint, b: bigint) => booleannewer: (a: biginta, b: bigintb) => a: biginta > b: bigintb,
              key: (id: bigint) => stringkey: var String: StringConstructorString,
            });
          }).Pipeable.pipe<Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, SqlError, Scope>, Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, never, Scope>>(this: Effect.Effect<...>, ab: (_: Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, SqlError, Scope>) => Effect.Effect<Stream.Stream<TrackedEnvelope<unknown>, never, never>, never, Scope>): Effect.Effect<...> (+21 overloads)pipe(import EffectEffect.const orDie: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, never, R>orDie),
        ),
    };
    return const shape: {
    publish: (channel: string, data: unknown) => Effect.Effect<string, never, never>;
    subscribe: (channel: string, o?: {
        readonly lastEventId?: string | undefined;
    }) => Stream.Stream<TrackedEnvelope<unknown>, never, never>;
}shape as interface PublisherShape<TEvents extends PublisherEvents>PublisherShape<function (type parameter) TEvents in <TEvents extends PublisherEvents>(): Effect.Effect<PublisherShape<TEvents>, never, PgClient.PgClient>TEvents>;
  });

const const layer: Layer.Layer<Posts, SqlError, never>layer = import LayerLayer.const effect: <Posts, PublisherShape<PostEvents>>(service: Key<Posts, PublisherShape<PostEvents>>) => <E, R>(effect: Effect.Effect<PublisherShape<PostEvents>, E, R>) => Layer.Layer<Posts, E, Exclude<R, Scope>> (+1 overload)effect(class PostsPosts)(const pgPublisher: <PostEvents>() => Effect.Effect<PublisherShape<PostEvents>, never, PgClient.PgClient>pgPublisher<type PostEvents = {
    created: Post;
    deleted: {
        id: string;
    };
}PostEvents>()).Pipeable.pipe<Layer.Layer<Posts, never, PgClient.PgClient>, Layer.Layer<Posts, SqlError, never>>(this: Layer.Layer<Posts, never, PgClient.PgClient>, ab: (_: Layer.Layer<Posts, never, PgClient.PgClient>) => Layer.Layer<Posts, SqlError, never>): Layer.Layer<Posts, SqlError, never> (+21 overloads)pipe(
  import LayerLayer.const provide: <never, SqlError, PgClient.PgClient | SqlClient>(that: Layer.Layer<PgClient.PgClient | SqlClient, SqlError, never>) => <RIn2, E2, ROut2>(self: Layer.Layer<ROut2, E2, RIn2>) => Layer.Layer<ROut2, SqlError | E2, Exclude<RIn2, PgClient.PgClient | SqlClient>> (+3 overloads)provide(
    import PgClientPgClient.const layer: (config: PgClient.PgPoolConfig) => Layer.Layer<PgClient.PgClient | SqlClient, SqlError>layer({ PgClientConfig.url?: Redacted.Redacted<string> | undefinedurl: import RedactedRedacted.const make: <string>(value: string, options?: {
    readonly label?: string | undefined;
}) => Redacted.Redacted<string>make(var process: NodeJS.Processprocess.NodeJS.Process.env: NodeJS.ProcessEnvenv.string | undefinedDATABASE_URL!) }),
  ),
);

Credits

These backends follow the adapters of oRPC's publisher.