Publisher | tRPC

Publisher is typed publish/subscribe as an Effect service. Declare a key with its channels, then provide a layer:

import { const Publisher: {
    Service: <Self, TEvents extends PublisherEvents>() => <const Key extends string>(key: Key) => PublisherClass<Self, Key, TEvents>;
}Publisher } from '@trpc12/trpc/server';

class class PostsPosts extends const Publisher: {
    Service: <Self, TEvents extends PublisherEvents>() => <const Key extends string>(key: Key) => PublisherClass<Self, Key, TEvents>;
}Publisher.type Service: <Posts, {
    created: Post;
    deleted: {
        id: string;
    };
}>() => <const Key extends string>(key: Key) => PublisherClass<Posts, Key, {
    created: Post;
    deleted: {
        id: string;
    };
}>Service<
  class PostsPosts,
  { created: Postcreated: Post; deleted: {
    id: string;
}deleted: { id: stringid: string } }
>()('Posts') {}

const const router: {
    create: BuildProcedure<InputDef<RootDef<{}, {}>, z.ZodObject<{
        title: z.ZodString;
    }, z.core.$strip>>, "mutation", {
        output: Post;
        errors: never;
        services: Posts;
        unmapped: never;
    }>;
    onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Post>;
        errors: never;
        services: Posts;
    }>;
}router = {
  create: BuildProcedure<InputDef<RootDef<{}, {}>, z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>>, "mutation", {
    output: Post;
    errors: never;
    services: Posts;
    unmapped: never;
}>create: const t: TrpcRoot<{}, {}, RootDef<{}, {}>>t.TrpcRoot<{}, {}, RootDef<{}, {}>>.procedure: ProcedureBuilder<RootDef<{}, {}>>procedure.ProcedureBuilder<RootDef<{}, {}>>.input<z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>>(schema: z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>): ProcedureBuilder<InputDef<RootDef<{}, {}>, z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>>> (+1 overload)input(const PostInput: z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>PostInput).ProcedureBuilder<InputDef<RootDef<{}, {}>, ZodObject<{ title: ZodString; }, $strip>>>.mutation<Effect.Effect<Post, never, Posts>>(resolver: (opts: ResolverOpts<InputDef<RootDef<{}, {}>, z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>>>) => Effect.Effect<Post, never, Posts>): BuildProcedure<InputDef<RootDef<{}, {}>, z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>>, "mutation", {
    output: Post;
    errors: never;
    services: Posts;
    unmapped: never;
}>mutation((opts: ResolverOpts<InputDef<RootDef<{}, {}>, z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>>>opts) =>
    import EffectEffect.const gen: <Effect.Effect<Post, never, never> | Effect.Effect<string, never, Posts>, Post>(f: () => Generator<Effect.Effect<Post, never, never> | Effect.Effect<string, never, Posts>, Post, never>) => Effect.Effect<Post, never, Posts> (+1 overload)gen(function* () {
      const const post: Postpost = yield* function savePost(input: {
    title: string;
}): Effect.Effect<Post>savePost(opts: ResolverOpts<InputDef<RootDef<{}, {}>, z.ZodObject<{
    title: z.ZodString;
}, z.core.$strip>>>opts.ResolverOpts<InputDef<RootDef<{}, {}>, ZodObject<{ title: ZodString; }, $strip>>>.input: {
    title: string;
}input);
      yield* class PostsPosts.PublisherClass<Posts, "Posts", { created: Post; deleted: { id: string; }; }>.publish<"created">(channel: "created", data: Post): Effect.Effect<string, never, Posts>publish('created', const post: Postpost);
      return const post: Postpost;
    }),
  ),
  onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
    output: TrackedEnvelope<Post>;
    errors: never;
    services: Posts;
}>onCreated: const t: TrpcRoot<{}, {}, RootDef<{}, {}>>t.TrpcRoot<{}, {}, RootDef<{}, {}>>.procedure: ProcedureBuilder<RootDef<{}, {}>>procedure.ProcedureBuilder<RootDef<{}, {}>>.subscription<Stream<TrackedEnvelope<Post>, never, Posts>>(resolver: (opts: SubscriptionResolverOpts<RootDef<{}, {}>>) => Stream<TrackedEnvelope<Post>, never, Posts>): BuildProcedure<RootDef<{}, {}>, "subscription", {
    output: TrackedEnvelope<Post>;
    errors: never;
    services: Posts;
}>subscription((opts: SubscriptionResolverOpts<RootDef<{}, {}>>opts) =>
    class PostsPosts.PublisherClass<Posts, "Posts", { created: Post; deleted: { id: string; }; }>.subscribe<"created">(channel: "created", opts?: SubscribeOptions): Stream<TrackedEnvelope<Post>, never, Posts>subscribe('created', { SubscribeOptions.lastEventId?: string | undefinedlastEventId: opts: SubscriptionResolverOpts<RootDef<{}, {}>>opts.SubscriptionResolverOpts<RootDef<{}, {}>>.lastEventId: string | undefinedlastEventId }),
  ),
};

createFetchHandler<{
    create: BuildProcedure<InputDef<RootDef<{}, {}>, z.ZodObject<{
        title: z.ZodString;
    }, z.core.$strip>>, "mutation", {
        output: Post;
        errors: never;
        services: Posts;
        unmapped: never;
    }>;
    onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Post>;
        errors: never;
        services: Posts;
    }>;
}, undefined>(opts: FetchHandlerOptions<{
    create: BuildProcedure<InputDef<RootDef<{}, {}>, z.ZodObject<{
        title: z.ZodString;
    }, z.core.$strip>>, "mutation", {
        output: Post;
        errors: never;
        services: Posts;
        unmapped: never;
    }>;
    onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Post>;
        errors: never;
        services: Posts;
    }>;
}, undefined>): FetchHandler<...>createFetchHandler({ BaseHandlerOptions<{ create: BuildProcedure<InputDef<RootDef<{}, {}>, ZodObject<{ title: ZodString; }, $strip>>, "mutation", { output: Post; errors: never; services: Posts; unmapped: never; }>; onCreated: BuildProcedure<...>; }>.router: {
    create: BuildProcedure<InputDef<RootDef<{}, {}>, z.ZodObject<{
        title: z.ZodString;
    }, z.core.$strip>>, "mutation", {
        output: Post;
        errors: never;
        services: Posts;
        unmapped: never;
    }>;
    onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Post>;
        errors: never;
        services: Posts;
    }>;
}router, layer: Layer<Posts, any, never>layer: class PostsPosts.PublisherClass<Posts, "Posts", { created: Post; deleted: { id: string; }; }>.memory(opts?: MemoryPublisherOptions): Layer<Posts, never, never>memory({ MemoryPublisherOptions.buffer?: number | undefinedbuffer: 100 }) });

subscribe() yields tracked() events with the publisher's ids, so reconnects resume. Posts.memory({ buffer }) serves one process on a PubSub and keeps buffer events per channel for resumes. For several processes, provide another layer with the same shape (publish, subscribe), built on the clients Effect already ships (Backends).

A channel per audience

To reach some subscribers only (a room, a user's inbox), give each audience its own channel. A template literal key types every one of them; take the subscriber's channel from ctx on the server, never from client input:

class class NotesNotes extends const Publisher: {
    Service: <Self, TEvents extends PublisherEvents>() => <const Key extends string>(key: Key) => PublisherClass<Self, Key, TEvents>;
}Publisher.type Service: <Notes, {
    [inbox: `inbox:${string}`]: {
        text: string;
    };
}>() => <const Key extends string>(key: Key) => PublisherClass<Notes, Key, {
    [inbox: `inbox:${string}`]: {
        text: string;
    };
}>Service<
  class NotesNotes,
  { [inbox: `inbox:${string}`inbox: `inbox:${string}`]: { text: stringtext: string } }
>()('Notes') {}

const const router: {
    notify: BuildProcedure<InputDef<RootDef<{
        user: User;
    }, {}>, Schema.Struct<{
        readonly to: Schema.String;
        readonly text: Schema.String;
    }>>, "mutation", {
        output: string;
        errors: never;
        services: Notes;
        unmapped: never;
    }>;
    inbox: BuildProcedure<RootDef<{
        user: User;
    }, {}>, "subscription", {
        output: TrackedEnvelope<{
            text: string;
        }>;
        errors: never;
        services: Notes;
    }>;
}router = {
  notify: BuildProcedure<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>, "mutation", {
    output: string;
    errors: never;
    services: Notes;
    unmapped: never;
}>notify: const t: TrpcRoot<{
    user: User;
}, {}, RootDef<{
    user: User;
}, {}>>t.TrpcRoot<{ user: User; }, {}, RootDef<{ user: User; }, {}>>.procedure: ProcedureBuilder<RootDef<{
    user: User;
}, {}>>procedure
    .ProcedureBuilder<RootDef<{ user: User; }, {}>>.input<Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>(schema: Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>): ProcedureBuilder<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>> (+1 overload)input(import SchemaSchema.function Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>(fields: {
    readonly to: Schema.String;
    readonly text: Schema.String;
}): Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>Struct({ to: Schema.Stringto: import SchemaSchema.const String: Schema.StringString, text: Schema.Stringtext: import SchemaSchema.const String: Schema.StringString }))
    .ProcedureBuilder<InputDef<RootDef<{ user: User; }, {}>, Struct<{ readonly to: String; readonly text: String; }>>>.mutation<Effect<string, never, Notes>>(resolver: (opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>>) => Effect<string, never, Notes>): BuildProcedure<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>, "mutation", {
    output: string;
    errors: never;
    services: Notes;
    unmapped: never;
}>mutation((opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>>opts) =>
      class NotesNotes.PublisherClass<Notes, "Notes", { [inbox: `inbox:${string}`]: { text: string; }; }>.publish<`inbox:${string}`>(channel: `inbox:${string}`, data: {
    text: string;
}): Effect<string, never, Notes>publish(`inbox:${opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>>opts.ResolverOpts<InputDef<RootDef<{ user: User; }, {}>, Struct<{ readonly to: String; readonly text: String; }>>>.input: {
    readonly to: string;
    readonly text: string;
}input.to: stringto}`, { text: stringtext: opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly to: Schema.String;
    readonly text: Schema.String;
}>>>opts.ResolverOpts<InputDef<RootDef<{ user: User; }, {}>, Struct<{ readonly to: String; readonly text: String; }>>>.input: {
    readonly to: string;
    readonly text: string;
}input.text: stringtext }),
    ),
  inbox: BuildProcedure<RootDef<{
    user: User;
}, {}>, "subscription", {
    output: TrackedEnvelope<{
        text: string;
    }>;
    errors: never;
    services: Notes;
}>inbox: const t: TrpcRoot<{
    user: User;
}, {}, RootDef<{
    user: User;
}, {}>>t.TrpcRoot<{ user: User; }, {}, RootDef<{ user: User; }, {}>>.procedure: ProcedureBuilder<RootDef<{
    user: User;
}, {}>>procedure.ProcedureBuilder<RootDef<{ user: User; }, {}>>.subscription<Stream<TrackedEnvelope<{
    text: string;
}>, never, Notes>>(resolver: (opts: SubscriptionResolverOpts<RootDef<{
    user: User;
}, {}>>) => Stream<TrackedEnvelope<{
    text: string;
}>, never, Notes>): BuildProcedure<RootDef<{
    user: User;
}, {}>, "subscription", {
    output: TrackedEnvelope<{
        text: string;
    }>;
    errors: never;
    services: Notes;
}>subscription((opts: SubscriptionResolverOpts<RootDef<{
    user: User;
}, {}>>opts) =>
    class NotesNotes.PublisherClass<Notes, "Notes", { [inbox: `inbox:${string}`]: { text: string; }; }>.subscribe<`inbox:${string}`>(channel: `inbox:${string}`, opts?: SubscribeOptions): Stream<TrackedEnvelope<{
    text: string;
}>, never, Notes>subscribe(`inbox:${opts: SubscriptionResolverOpts<RootDef<{
    user: User;
}, {}>>opts.ResolverOpts<RootDef<{ user: User; }, {}>>.ctx: {
    user: User;
}ctx.user: Useruser.User.id: stringid}`, {
      SubscribeOptions.lastEventId?: string | undefinedlastEventId: opts: SubscriptionResolverOpts<RootDef<{
    user: User;
}, {}>>opts.SubscriptionResolverOpts<RootDef<{ user: User; }, {}>>.lastEventId: string | undefinedlastEventId,
    }),
  ),
};

The backend filters by channel, so a process only receives its subscribers' channels (Redis subscribes per channel). To reach several audiences, publish to each channel.

A channel per audience

Pages under Publisher