> ## Documentation Index
> Fetch the complete documentation index at: https://docs.myme.so/llms.txt
> Use this file to discover all available pages before exploring further.

# Sync engine

> SwiftUI reactive queries, sync engine, sync events, end-to-end subscription, local IDs

## SwiftUI reactive queries

On a pure-local or synced client, `client.makeStore()` returns a `MarfaStore` — a `@MainActor @Observable` factory for live queries driven by `ModelContext.didSave` with a short debounce window and a `FetchDescriptor` refetch per save.

```swift theme={null}
import SwiftUI
import MarfaSDK

struct NotesView: View {
  @State private var store: MarfaStore?
  @State private var query: ItemQuery?

  var body: some View {
    List(query?.items ?? [], id: \.id) { item in
      Text(item.properties["body"]?.stringValue ?? "")
    }
    .task {
      let client = try? await MarfaClient.local(path: dbPath)
      self.store = client?.makeStore()
      self.query = store?.query(filters: ListFilters(type: "core.note"))
    }
  }
}
```

The query classes:

* `ItemQuery` — `[Item]` matching a `ListFilters`
* `TypedItemQuery<T: MarfaItem>` — `[T]` — e.g. `TypedItemQuery<CoreNote>`
* `SingleItemQuery` — one item by id; `nil` when purged
* `ItemsWithMetadataQuery` — `[ItemWithMetadata]` pairs — item plus its metadata row. Re-runs when either the items or item-metadata tables change. Use when a view needs per-item tags or favorite flags alongside the item (replaces hand-rolling a cache subscribed to `SyncEngine.events`).
* `EdgesQuery` — outbound edges for a `sourceId`, optionally filtered by `edgeType`. A second initialiser takes only an `edgeType` to track **all edges of a type** across the local store — `store.queryEdges(ofType: "in-thread")` for taxonomy-style "every reply" surfaces.
* `TagsQuery` — `[TagWithCount]` for the whole tenant. Re-runs the tag aggregation when items or metadata change.
* `BackrefsQuery` — `[String: [Edge]]` keyed by target id. Re-runs when edges change.
* `PendingMutationsQuery` — `[PendingMutationSummary]` for the mutation queue. See [Pending mutations](#pending-mutations).
* `BlobUploadProgressQuery` — `[String: BlobUploadProgress]` keyed by content hash. See [Blob upload progress](#blob-upload-progress).

All are `@Observable @MainActor`. Views observe them directly — no `ObservableObject`, no manual Combine plumbing. Each query exposes its data property (`items`, `item`, `edges`, `tags`, or `edgesByTarget`), `isLoading`, and `error`. Call `stop()` to tear down the underlying observation.

```swift theme={null}
struct TagCloud: View {
  @State private var tags: TagsQuery?
  @State private var replies: BackrefsQuery?

  var body: some View {
    VStack {
      ForEach(tags?.tags ?? [], id: \.tag) { entry in
        Text("\(entry.tag) (\(entry.count))")
      }
    }
    .task {
      tags = store.queryTags()
      replies = store.queryBackrefs(to: messageIds, edgeType: "in-thread")
    }
  }
}
```

For views that render each item with its tags or favorite flag, use `ItemsWithMetadataQuery`:

```swift theme={null}
struct NoteFeed: View {
  @State private var query: ItemsWithMetadataQuery?

  var body: some View {
    List(query?.items ?? [], id: \.item.id) { pair in
      NoteRow(
        item: pair.item,
        tags: pair.metadata.tags,
        isFavorite: pair.metadata.tags.contains("favorite")
      )
    }
    .task {
      query = store.queryItemsWithMetadata(filters: .init(type: "core.note"))
    }
  }
}
```

<Note>
  `TagsQuery` and `ItemsWithMetadataQuery` both re-run their aggregations on every relevant write. They're tuned for UI-scale surfaces, not unbounded analytics.
</Note>

Network-only clients return `nil` from `makeStore()` — the reactive layer only materialises when a local store is configured.

### Pending mutations

`store.queryPendingMutations()` vends a `PendingMutationsQuery` over the local mutation queue — every write that's been enqueued but not yet replayed to the server. Use it to render per-item "syncing" badges, queue-depth banners, or retry toasts without subscribing to `SyncEngine.events` and reimplementing the projection.

```swift theme={null}
struct SyncStatusBar: View {
  @State private var query: PendingMutationsQuery?

  var body: some View {
    Group {
      if let q = query, !q.isEmpty {
        HStack {
          ProgressView()
          Text("\(q.mutations.count) pending")
        }
      }
    }
    .task {
      query = store?.queryPendingMutations()
    }
  }
}
```

Each entry is a `PendingMutationSummary` with `id`, `kind` (e.g. `.createItem`, `.uploadBlob`, `.setMetadata`), `itemId`, `createdAt`, and a projected `status`:

* `.pending` — queued, not yet attempted.
* `.inFlight` — the sync engine is issuing the transport call right now.
* `.retrying(attemptCount: Int, lastError: String)` — a transient failure, will retry on the next drain cycle.

The query refreshes on every `ModelContext.didSave` — enqueues, in-flight transitions, retries, and removals all land on the same observation path as every other reactive query. Returns `nil` on clients without a mutation queue (network-only mode).

### Blob upload progress

`store.queryBlobUploadProgress()` vends a `BlobUploadProgressQuery` that projects the four `.blobUpload*` sync events into a dict keyed by content hash. Use it for global upload HUDs or per-attachment progress bars without reading the raw event stream.

```swift theme={null}
struct AttachmentProgressView: View {
  @State private var query: BlobUploadProgressQuery?
  let hash: String

  var body: some View {
    Group {
      if let entry = query?.uploads[hash] {
        switch entry.state {
        case .uploading(let sent, let total):
          ProgressView(value: Double(sent), total: Double(total))
        case .failed:
          Label("Upload failed", systemImage: "exclamationmark.triangle")
        case .pending, .completed:
          EmptyView()
        }
      }
    }
    .task {
      query = store?.queryBlobUploadProgress()
    }
  }
}
```

Entries are evicted on `.completed` — apps wanting a "recently completed" fade should snapshot the final value in their own view layer. `.failed` entries linger so a consumer can render a retry affordance; a subsequent transient retry overwrites the entry with a fresh `.uploading(0, total)`. Returns `nil` on clients without a sync engine (network-only and pure-local modes).

## Sync engine

In synced mode, `client.syncEngine` is non-nil. Start it when your app is ready to sync:

```swift theme={null}
await client.syncEngine?.start()
```

The engine:

1. Watches `NWPathMonitor` via `ConnectionStateManager` and opens an SSE stream on `GET /events` when the path goes online.
2. Applies SSE events into the local store. Persists `Last-Event-ID` so reconnection resumes from the correct point.
3. Drains the `MutationQueue` both when the stream is idle and within a short debounce window of any new enqueue, replaying local writes against the server. The mutation queue captures per-call `UpdateOptions.conflict` so the chosen strategy applies on replay; the `.callback` resolver closure is call-site-only and degrades to `.auto` on replay because closures aren't serialisable.
4. On failure, records the attempt against the queue record and retries on the next reachability cycle.

`stop()` tears the engine down gracefully.

#### Proactive drain

The engine listens on `MutationQueue.drainRequests` and schedules a replay within `drainDebounceInterval` of each enqueue while the connection is online — a device that's online but idle no longer holds pending writes until the next SSE reconnect. Offline, connecting, and syncing states short out: the queue sits silently, the post-SSE-close drain picks up, or an in-flight drain coalesces the burst respectively. Tune via `SyncEngine.init(drainDebounceInterval:)` (default `.milliseconds(150)`).

### Sync events

Subscribe to `client.syncEngine?.events` for typed sync activity:

```swift theme={null}
for await event in client.syncEngine?.events ?? .init({ _ in }) {
  switch event {
  case .synced(let at):                          // every successful drain
  case .failed(let error):                       // last error during a drain
  case .conflictAutoMerged(let payload):         // replay auto-merged a 409
  case .itemCreated(let id):
  case .itemUpdated(let id):
  case .itemDeleted(let id):
  case .edgeCreated(let id):
  case .edgeDeleted(let id):
  case .mutationDropped(let kind, let itemId,
                        let attempt, let error): // permanent failure; queue record removed
  case .blobUploadStarted(let hash, let total):
  case .blobUploadProgress(let hash, let sent, let total):
  case .blobUploadCompleted(let hash):
  case .blobUploadFailed(let hash, let error):
  }
}
```

The four `.blobUpload*` cases are the raw signal that [`BlobUploadProgressQuery`](#blob-upload-progress) projects into its observable dict — most apps subscribe to the query, not the events directly. `.mutationDropped` fires when a queued mutation exhausts its retries and is removed permanently.

Replaces the pattern of inferring sync state from `ConnectionState` transitions. Multi-subscriber — each access to `events` returns a fresh stream; past events are not replayed to late subscribers.

#### End-to-end subscription

Own the lifecycle from a single place — typically an `@Observable` service or a SwiftUI `.task { }` — so the subscriber task cancels when the view goes away and the sync engine stops when the app backgrounds.

```swift theme={null}
import MarfaSDK
import Observation

@MainActor
@Observable
final class SyncActivity {
  private(set) var lastSyncedAt: Date?
  private(set) var lastError: Error?
  private(set) var pendingConflictedCopies: [String] = []

  private var client: MarfaClient?
  private var task: Task<Void, Never>?

  // 1. Create — on app launch / sign-in.
  func start(client: MarfaClient) async {
    self.client = client
    await client.syncEngine?.start()

    // 2. Subscribe — each access to `events` is a fresh stream.
    task = Task { [weak self] in
      guard let events = client.syncEngine?.events else { return }
      for await event in events {
        guard !Task.isCancelled else { break }
        await self?.handle(event)
      }
    }
  }

  // 3. Handle — side-effect into observable state.
  private func handle(_ event: SyncEvent) {
    switch event {
    case .synced(let at):
      lastSyncedAt = at
      lastError = nil
    case .failed(let error):
      lastError = error
    case .conflictAutoMerged(let payload):
      if let sibling = payload.conflictedCopyId {
        pendingConflictedCopies.append(sibling)
      }
    case .itemCreated, .itemUpdated, .itemDeleted,
         .edgeCreated, .edgeDeleted, .mutationDropped,
         .blobUploadStarted, .blobUploadProgress,
         .blobUploadCompleted, .blobUploadFailed:
      break   // SwiftData-backed MarfaStore queries pick these up automatically
    }
  }

  // 4. Tear down — on sign-out / app termination.
  func stop() async {
    task?.cancel()
    task = nil
    await client?.syncEngine?.stop()
    client = nil
  }
}
```

From SwiftUI, scope the subscription to the view that needs it:

```swift theme={null}
struct RootView: View {
  @State private var activity = SyncActivity()
  let client: MarfaClient

  var body: some View {
    ContentView(activity: activity)
      .task {
        await activity.start(client: client)
      }
      // .task's implicit cancellation on disappearance ends the subscriber
      // loop via Task.isCancelled; call activity.stop() on explicit sign-out
      // to also tear the engine down.
  }
}
```

The SDK's reactive queries (`ItemQuery`, `TagsQuery`, `BackrefsQuery`) already update when SSE writes land in the local store, so most views don't need to observe `events` directly. Subscribe when you need *sync-lifecycle* side effects — a "last synced" footer, a conflict toast, a sign-out-on-auth-failure hook.

See [Realtime](/api/realtime) for the underlying SSE contract, event catalog, and `Last-Event-ID` semantics.

### Local IDs (UUIDv7)

`LocalStore.createItem` mints a UUIDv7 by default — RFC 9562, timestamp-prefixed, globally unique. Pass `CreateItemInput.id` explicitly to override (still encouraged for callers that need to reference the new id before `createItem` returns; both paths produce server-compatible IDs).