Pingline M2 — Realtime as an AsyncSequence

Supabase Realtime hands you database changes over a websocket via a callback-shaped API. The modern Swift move is to bridge that into an AsyncStream so the rest of your code consumes messages with for await — no delegate, no callback pyramid, and a clean cancellation story. This milestone builds that bridge and wires it to the UI.

Bridging the channel to a stream

Wrap the subscription in an AsyncStream. The stream's closure gets a continuation; you yield each incoming message into it and finish() when the subscription ends. The callback lives inside the closure, so the ugly part is contained in one place.

import Foundation
import Supabase

func messageStream(for conversationID: UUID) -> AsyncStream<Message> {
    AsyncStream { continuation in
        let channel = supabase.realtimeV2.channel("messages:\(conversationID)")

        let subscription = channel.onPostgresChange(
            InsertAction.self,
            schema: "public",
            table: "messages",
            filter: "conversation_id=eq.\(conversationID)"
        ) { action in
            if let message = try? action.decodeRecord(as: Message.self, decoder: apiDecoder) {
                continuation.yield(message)
            }
        }

        Task { await channel.subscribe() }

        // When the consumer stops iterating, tear the channel down.
        continuation.onTermination = { _ in
            Task { await channel.unsubscribe() }
        }
    }
}

Two things make this correct. First, continuation.yield(message) pushes each decoded insert to whoever is for await-ing the stream. Second — and this is the part people forget — onTermination unsubscribes when the consumer goes away (the view disappears, the task is cancelled). Without it you leak a live socket subscription every time the user backs out of a thread.

Consuming it in the view model

The view model is @Observable and lives on the @MainActor — it feeds SwiftUI, so its published state must mutate on the main actor. It kicks off a task that loops over the stream and appends each message.

import Foundation

@MainActor
@Observable
final class ThreadViewModel {
    private(set) var messages: [Message] = []
    private let conversationID: UUID
    private var receiveTask: Task<Void, Never>?

    init(conversationID: UUID) {
        self.conversationID = conversationID
    }

    func start() {
        receiveTask = Task { [conversationID] in
            for await incoming in messageStream(for: conversationID) {
                insert(incoming)
            }
        }
    }

    func stop() {
        receiveTask?.cancel()
        receiveTask = nil
    }

    private func insert(_ message: Message) {
        // Ignore a duplicate id (e.g. the echo of our own optimistic send).
        guard !messages.contains(where: { $0.id == message.id }) else { return }
        messages.append(message)
        messages.sort { $0.createdAt < $1.createdAt }
    }
}

The for await loop runs until the stream finishes or the task is cancelled. Because ThreadViewModel is @MainActor, insert mutates messages on the main actor, and SwiftUI picks up the change automatically. insert also dedupes by id — that guard is what lets the server's echo of your own message land harmlessly in Milestone 3.

Lifecycle in the view

Start the stream when the thread appears; cancel it when it leaves. SwiftUI's .task modifier does both — it starts the work on appear and cancels it automatically on disappear.

struct ThreadView: View {
    @State private var model: ThreadViewModel

    var body: some View {
        List(model.messages) { message in
            MessageRow(message: message)
        }
        .task { model.start() }   // auto-cancelled when the view goes away
    }
}

Checkpoint: your app should now…

• Subscribe to a conversation's inserts and expose them as an AsyncStream<Message>.
• Drive a @MainActor @Observable view model that appends and timestamp-sorts messages.
• Show a live transcript: a row inserted by another client appears in the open thread with no manual refresh, and backing out of the thread unsubscribes.

Stretch goals

• Fetch the recent history once on open, then merge the live stream on top of it.
• Surface ConnectionStatus changes as a second stream and show a "Reconnecting…" banner.

Knowledge check

Q: Why bridge the Realtime callback into an AsyncStream instead of using it directly? So the rest of the app consumes messages with for await, gets structured cancellation for free, and keeps the callback confined to one wrapper. The consuming code reads like sequential logic instead of nested callbacks.

Q: What breaks if you omit continuation.onTermination? The websocket subscription is never torn down when the consumer stops, so every time the user opens and closes a thread you leak a live channel. onTermination is where you unsubscribe.