Reactive reads¶

selectAsFlow reads a query as it changes: it emits the rows, with the user's queued writes laid over them, and again whenever a write, a refresh or a server record changes them. It needs no Realtime channel: it follows the device's cache, and refreshes it from the server as its strategy says.

From the samples: OfflineTaskRepository.kt · TaskRepository.swift · TasksModel.swift

AndroidiOS

/** The user's tasks that [query] selects, as a live read mapped to the app's snapshot. */
private fun taskList(
    query: TaskQuery,
    strategy: FlowReadStrategy,
    shown: (List<Task>) -> List<Task>,
): Flow<TaskListSnapshot> {
    val recent = recentWindow(clock.now()) // read once: the request must not read the clock
    return whileSignedIn(before = nothingYet()) {
        combine(
            offline.from(Tables.TASKS).selectAsFlow(readStrategy = strategy) {
                tasksMatching(query, recent)
            },
            pendingRowIds(Tables.TASKS),
        ) { snapshot, pending ->
            snapshot.toRowsSnapshot { result ->
                // A queued task is listed after the server's rows; show it where it sorts.
                shown(result.decodeList<TaskRow>().map { it.toTask(pending) }.newestFirst())
            }
        }
    }
}

/// Every task of the user, newest first, as the device holds it and as the server refreshes
/// it: the list the other screens share. The task screen picks its task out of it, because
/// each query is cached on its own, and offline a query never fetched has nothing to show.
func taskSnapshots(readStrategy: SequenceReadStrategy = .cacheFirstThenNetwork) -> QuerySnapshotSequence<[TaskRow]> {
    client.from(Tables.tasks)
        .selectAsSequence(as: [TaskRow].self, readStrategy: readStrategy)
        .order("created_at", ascending: false)
}

/// Follows every task of the signed-in user, the list the other screens share, until the
/// calling task is canceled. Before there is a user it asks nothing: the schema refuses every
/// request without one.
func observeTasks() async {
    rows = nil
    guard repository.session.userID != nil else { return }
    while !Task.isCancelled {
        do {
            for try await snapshot in repository.taskSnapshots() {
                rows = snapshot.result
            }
            return
        } catch is CancellationError {
            return
        } catch {
            // The shown list reports a failure; both read the same rows. Read again when it is
            // worth it, so the list does not stop.
            await untilWorthReadingAgain()
        }
    }
}

The three strategies¶

The sample's list takes the first two from its Prefer fresh data switch, and a task's photos take the third while Simulate offline is on:

From the samples: OfflineTaskRepository.kt · TaskRepository.swift · TaskDetailModel.swift

AndroidiOS

/** The list's live read, with the strategy the "Prefer fresh data" switch picks. */
@OptIn(ExperimentalCoroutinesApi::class)
override fun tasks(query: TaskQuery): Flow<TaskListSnapshot> =
    switches
        .map { it.preferFreshData }
        .distinctUntilChanged()
        .flatMapLatest { preferFresh ->
            val strategy =
                if (preferFresh) FlowReadStrategy.NetworkFirstThenCache // ask first
                else FlowReadStrategy.CacheFirstThenNetwork // the default: the device first
            taskList(query, strategy) { it }
        }


/**
 * A task's photo rows, from the query of [allPhotos]. While Simulate offline is on, the read
 * takes the device's copy alone and sends nothing; the screen says so.
 */
@OptIn(ExperimentalCoroutinesApi::class)
private fun storedPhotos(taskId: String): Flow<PhotoListSnapshot> =
    offline.syncStatus
        .map { it.network == NetworkStatus.DISABLED }
        .distinctUntilChanged()
        .flatMapLatest { switchedOff ->
            val strategy =
                if (switchedOff) FlowReadStrategy.CacheOnly
                else FlowReadStrategy.CacheFirstThenNetwork
            photoList(strategy, page = null) { photos ->
                photos.filter { it.taskId == taskId }.sortedBy { it.createdAt }
            }
        }

/// How the task list reads, as Settings → Prefer fresh data says: the device's rows at once,
/// refreshed from the server alongside (the default), or the server's answer first, the
/// device's rows only when the server cannot be reached.
static func readStrategy(prefersFreshData: Bool) -> SequenceReadStrategy {
    prefersFreshData ? .networkFirstThenCache : .cacheFirstThenNetwork
}

// Simulate offline: the device's rows only, without a request the switch would refuse.
let readStrategy: SequenceReadStrategy = isSimulatingOffline ? .cacheOnly : .cacheFirstThenNetwork
for try await snapshot in repository.photoSnapshots(readStrategy: readStrategy) {
    photoRows = snapshot.result ?? []
}

CacheFirstThenNetwork, the default, emits the cached rows at once and asks the server in parallel. A refresh that fails leaves the rows on screen and is tried again for as long as the flow is collected. Use it for almost every screen.
NetworkFirstThenCache asks the server first, then emits the cached rows as they change. Offline with the query cached, it emits them at once and keeps asking until the server answers; with nothing cached, or when the server answers with an error, the flow fails.
CacheOnly never sends a request. For a query that is not cached it emits null rows, or only the queued writes the query admits, rather than failing.

What each emission says¶

Each emission is a QuerySnapshot, and its fields are true together:

Field	Meaning
`result`	The rows, as supabase-kt's `PostgrestResult`, with the user's queued writes applied. `null` while nothing is known for the query: nothing cached and no queued write it admits. A body of `[]` means the server returned no rows
`isFromCache`	Whether the rows are the device's copy rather than an answer this read had from the server: `true` until the read's refresh succeeds, always under `CacheOnly`, and again once a write marks them stale
`fetchedAt`	When the server sent the rows, by the device's clock; never the time of a local change applied to them
`hasPendingMutations`	Whether a queued write added, changed or removed a row shown
`refresh`	How the refresh behind the read stands; see Refresh lifecycle

result == null is a loading state; an empty body is an empty list. Next to rows already on screen, refresh says whether to show a note (Refresh lifecycle has the sample's mapping). To mark each row whose change has not reached the server yet, combine the read with pendingMutations(table) and its rows' keys:

From the samples: OfflineTaskRepository.kt · TasksModel.swift

AndroidiOS

/** The ids of [table]'s rows a queued write touches, from each write's row keys. */
private fun pendingRowIds(table: String): Flow<Set<String>> =
    offline.pendingMutations(table).map { mutations -> mutations.rowIds() }

/// Follows which tasks have a write waiting in the mutation queue.
func observePendingTasks() async {
    do {
        for try await mutations in repository.pendingMutations(table: Tables.tasks) {
            pendingIDs = Set(mutations.flatMap(\.rowIDs))
        }
    } catch is CancellationError {
        return
    } catch {
        failures.report(error)
    }
}

Filters, search and pages¶

A live read takes supabase-kt's whole request DSL, and each request is cached on its own. The sample's chips, search and Recent chip are filters the server applies; a row's photo count reads the photos of the tasks shown with in; the Photos grid reads a page more each time:

From the samples: TaskQueries.kt · OfflineTaskRepository.kt · TaskQuery.swift · TaskRepository.swift

AndroidiOS

/** The chips as filters the server applies; the overlay evaluates `eq` and `is` offline too. */
private fun PostgrestFilterBuilder.chip(filter: TaskFilter, recent: RecentWindow) {
    when (filter) {
        TaskFilter.All -> Unit
        TaskFilter.Open -> eq(DONE, false) // done=eq.false
        TaskFilter.Done -> exact(DONE, true) // done=is.true
        TaskFilter.Recent -> madeIn(recent)
    }
}


/** Titles containing [text], ignoring case; a `%`, `_` or `\` in it matches only itself. */
private fun PostgrestFilterBuilder.titleContains(text: String) {
    val literal = text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    ilike(TITLE, "%$literal%")
}

/**
 * The photo rows of [taskIds]: one live read, by `in` on `task_id`. A set of tasks never read
 * online has no rows offline, so it is null for it.
 */
private fun photoRowsOf(taskIds: List<String>): Flow<List<PhotoTaskRow>?> {
    val ids = taskIds.sorted() // the same tasks, the same query, the same cached rows
    return whileSignedIn(before = null) {
        offline[Tables.TASK_PHOTOS] // the same builder as from(Tables.TASK_PHOTOS)
            .selectAsFlow(Columns.list(ID, TASK_ID)) { filter { isIn(TASK_ID, ids) } }
            .map { snapshot -> snapshot.result?.decodeList<PhotoTaskRow>() }
    }
}


/** The user's photos, newest first: all of them, or the first [page] when the grid pages. */
private fun photoList(
    strategy: FlowReadStrategy,
    page: Int?,
    shown: (List<TaskPhoto>) -> List<TaskPhoto>,
): Flow<PhotoListSnapshot> =
    whileSignedIn(before = nothingYet()) {
        combine(
            offline.from(Tables.TASK_PHOTOS).selectAsFlow(readStrategy = strategy) {
                newestFirst()
                if (page != null) range(0L, page - 1L) // each page size is a query of its own
            },
            pendingRowIds(Tables.TASK_PHOTOS),
        ) { snapshot, pending ->
            snapshot.toRowsSnapshot { result ->
                val photos = result.decodeList<TaskPhotoRow>().map { it.toTaskPhoto(pending) }
                shown(photos.sortedByDescending { it.createdAt })
            }
        }
    }

/// `request` narrowed to this chip's tasks, by the server.
func applied<Request: OfflinePostgrestFilterBuilder>(to request: Request, recentWindow: DateInterval) -> Request {
    switch self {
    case .all: request
    case .open: request.neq("done", value: true)
    case .done: request.is("done", value: true)
    case .recent: TaskQuery.created(within: recentWindow, request)
    }
}

/// The tasks whose title holds `text`, in any case: `ilike`, which the server applies.
static func titleContains<Request: OfflinePostgrestFilterBuilder>(_ text: String, _ request: Request) -> Request {
    request.ilike("title", pattern: "%\(text)%")
}

/// The photo rows of `taskIDs`, the tasks on screen, in one query: the list shows how many
/// photos each one has.
func photoSnapshots(ofTasks taskIDs: [String]) -> QuerySnapshotSequence<[TaskPhotoRow]> {
    client.from(Tables.taskPhotos)
        .selectAsSequence(as: [TaskPhotoRow].self)
        .in("task_id", values: taskIDs)
        .order("created_at", ascending: true)
}

/// The user's newest `count` photos, for the Photos grid, which asks for a page more each
/// time it reaches the end.
func photoPage(count: Int) -> QuerySnapshotSequence<[TaskPhotoRow]> {
    client.from(Tables.taskPhotos)
        .selectAsSequence(as: [TaskPhotoRow].self)
        .order("created_at", ascending: false)
        .limit(count)
}

A queued change to done moves a task between Open and Done at once, but a task made offline shows in every search until the list is fetched again (Optimistic overlay).

When it fetches again¶

A CacheFirstThenNetwork read refreshes when it starts and tries again until a refresh succeeds. After that it fetches again only when local data is cleared or the cache passes to another user, when a write the cache could not follow exactly marks the list stale, and when a new collection starts after every earlier collector has gone. It does not poll: to pick up changes made elsewhere, read the same query with readStrategy = ReadStrategy.NetworkOnly (a pull to refresh), whose answer replaces the cached list for every live reader of it, or apply Realtime records (Applying Realtime events).

Collectors of one CacheFirstThenNetwork query share one subscription to the cache and one refresh. A failure the read cannot recover from ends the flow after the values before it; a refresh that fails is not one, it is reported in refresh. When the client closes, a collection still running completes: first() on one that has not emitted yet throws NoSuchElementException (firstOrNull() returns null), and a flow collected after close() fails with IllegalStateException.

Behavior matrix¶

Facts about the engine, one row per condition; the conditions are those of the one-shot matrix.

Condition	`CacheFirstThenNetwork`	`NetworkFirstThenCache`	`CacheOnly`
Online	The cached payload with the overlay at once (`null` when nothing is cached), `Refreshing`; then the server's rows, `Idle`. Collectors of one query share one refresh.	The server's rows, `Idle`; then the cache as it changes.	The cached payload with the overlay, or `null`; `Idle`; never a request.
Offline, warm	The cached payload with the overlay. The refresh retries for the life of the collection (300 ms, 900 ms, 2.4 s, 5 s, 15 s, then every 60 s), `Stalled` after four failures, about 3.6 s: each attempt fails at once, without supabase-kt's retries. The flow stays live.	The cached payload with the overlay at once, `Stalled`: the cache is not confirmed. It fetches again on the schedule `CacheFirstThenNetwork` retries on, and is `Idle` once the server answers; the flow stays live.	The cached payload with the overlay.
Offline, cold	`null`, `Refreshing`, then `Stalled`; the flow stays live and settles when the server answers.	The flow fails with the connectivity failure.	`null`, or the queued rows the query admits.
Network known down (the monitor says offline, or `disableNetwork()`)	The cached payload, `Paused`, not `Stalled`: waiting for the network, not failing. While access is off nothing is sent; when the network comes back the refresh runs at once and is `Idle`.	The cached payload, from the cache, `Paused`.	As offline.
Session restoring	The cached payload. Nothing a request returns is written, and the refresh does not settle (it reads `Refreshing`), until the session has. Settled as the stored user: the user's answer. Settled signed out: a stored user's rows show while the session restores and are dropped when it settles signed out; with no stored session they are never shown. It then settles on the anonymous answer.	Nothing until the session settles, as a one-shot read waits; then the user's rows, `Idle`. A session that never settles: the cache, still `Refreshing`, since waiting for the session is not a failure.	The cached payload, no request.

A server's own error (4xx or 5xx) is a refresh failure for CacheFirstThenNetwork, and fails a NetworkFirstThenCache flow even with a warm cache.

A read of a shape the cache cannot hold (single(), a count, csv(), …; see the one-shot matrix) emits the network's answer once, Idle, and completes; offline it fails. Under CacheOnly it fails at once with CacheMissException and sends nothing.

A collector slow to take each value is not handed every list in between: under CacheFirstThenNetwork it gets the latest, under the other two a few superseded lists at most. It ends on the latest, and never holds back another collector of the same query.

A read live while the session settles into another identity drops what it showed for the one before and ends on the new identity's rows alone.

Each snapshot field is true of its own emission, as the table above says; a refresh that keeps failing while the network is believed up is Stalled.

Reactive reads¶

The three strategies¶

What each emission says¶

Filters, search and pages¶

When it fetches again¶

Sharing and closing¶

Behavior matrix¶