# Subscriptions

struct Resolver {
    // Example using an async generator
    func hello(_: Context, _: NoArguments) async -> EventStream<String> {
        let stream = AsyncStream { con in
            for word in ["Hello", "Bonjour", "Ciao"] {
                con.yield(word)
            }
            con.finish()
        }
        return stream.toEventStream()
    }

    func postCreated(ctx: Context, _: NoArguments) async -> EventStream<Post> {
        ctx.pubsub.asyncStream(Post.self, for: "POST_CREATED").toEventStrem()
    }
}

import Pioneer

let pubsub = AsyncPubSub()

await pubsub.publish("POST_CREATED", Post(by: "Jeff Bezos", content: "How much many do I have"))

struct Resolver {
    ...

    func createPost(ctx: Context, args: CreatePostArgs) async throws -> Post {
        let post = Post(args: args)
        try await post.create(on: ctx.req.db)
        await ctx.pubsub.publish(for: "POST_CREATED", post)
        return post
    }
}

pubsub.asyncStream(Post.self, for: "POST_CREATED");

struct Resolver {
    ...

    func postCreated(ctx: Context, _: NoArguments) async -> EventStream<Post> {
        ctx.pubsub.asyncStream(Post.self, for: "POST_CREATED").toEventStrem()
    }
}

import Pioneer
import Redis
import Foundation
import NIOFoundationCompat

struct RedisPubSub: PubSub {

    // MARK: - Actor for distribution
    actor Dispatcher {
        private let redis: RedisClient
        private var broadcasting: [String: Broadcast<Data>] = [:]

        init(_ redis: RedisClient) {
            self.redis = redis
        }

        /// Get a downstream from the broadcast for the channel given
        func downstream(to channel: String) async -> AsyncStream<Data> {
            let broadcast = await subscribe(to: channel)
            let downstream = await broadcast.downstream()
            return downstream.stream
        }

        /// Get the broadcast for the channel if exist, otherwise make a new one
        private func subscribe(to channel: String) async -> Broadcast<Data> {
            if let broadcast = broadcasting[channel] {
                return broadcast
            }
            let broadcast = Broadcast<Data>()
            broadcasting[channel] = broadcast
            await apply(from: .init(channel), to: broadcast)
            return broadcast
        }

        /// Apply broadcasting to the Redis channel subscription
        private func apply(from channel: RedisChannelName, to broadcast: Broadcast<Data>) async {
            do {
                try await redis.subscribe(
                    to: channel,
                    messageReceiver: { _, msg in
                        guard case .bulkString(.some(let buffer)) = msg else { return }
                        let data = Data(buffer: buffer)
                        Task {
                            await broadcast.publish(data)
                        }
                    },
                    onUnsubscribe: { _, _ in
                        Task {
                            await broadcast.close()
                        }
                    }
                )
                .get()
            } catch {
              await broadcast.close()
            }
        }

        /// Pubblish the data (which is RESPValueConvertible) to the specific redis channel
        func publish(for channel: String, _ value: Data) async {
            let _ = try? await redis.publish(value, to: .init(channel)).get()
        }

        /// Close the redis channel subscription and all of the downstreams
        func close(for channel: String) async {
            try? await redis.unsubscribe(from: .init(channel)).get()
            await broadcasting[channel]?.close()
        }
    }

    // MARK: -- Protocol required methods

    public func asyncStream<DataType: Sendable & Decodable>(_ type: DataType.Type = DataType.self, for trigger: String) -> AsyncStream<DataType> {
        AsyncStream<DataType> { con in
            let task = Task {
                let stream = await dispatcher.downstream(to: trigger)
                for await data in stream {
                    guard let event = try? JSONDecoder().decode(DataType.self, data) else { continue }
                    con.yield(event)
                }
                con.finish()
            }
            con.onTermination = { @Sendable _ in
                task.cancel()
            }
        }
    }

    public func publish<DataType: Sendable & Encodable>(for trigger: String, payload: DataType) async {
        guard let data = try? JSONEncoder().encode(payload) else { return }
        await dispatcher.publish(for: trigger, data)
    }

    public func close(for trigger: String) async {
        await dispatcher.close(for: trigger)
    }

    // MARK: - Properties

    private let dispatcher: Dispatcher

    public init(_ redis: RedisClient) {
        self.dispatcher = .init(redis)
    }
}

struct Message: Sendable, Codable { ... }

struct Resolver {
    let pubsub: PubSub = app.environment.isRelease ? RedisPubSub(app.redis) : AsyncPubSub()

    func create(ctx: Context, _: NoArguments) async -> Message {
        let message = ...
        await pubsub.publish(message)
        return message
    }

    func onCreate(ctx: Context, _: NoArguments) async -> EventStream<Message> {
        pubsub.asyncStream(Message.self, for: "message-create").toEventStream()
    }
}