Sync Model

// From packages/zero-cache/src/services/replicator/incremental-sync.ts
export class IncrementalSyncer {
  async run(lc: LogContext) {
    while (this.#state.shouldRun()) {
      const {replicaVersion, watermark} = getSubscriptionState(this.#replica);
      
      downstream = await this.#changeStreamer.subscribe({
        protocolVersion: PROTOCOL_VERSION,
        watermark,        // Resume from last processed position
        replicaVersion,
      });
      
      for await (const message of downstream) {
        // Process each replication message
        const result = processor.processMessage(lc, message);
        
        if (result?.watermark && result?.changeLogUpdated) {
          // Notify clients when new version is ready
          void this.#notifier.notifySubscribers({state: 'version-ready'});
        }
      }
    }
  }
}

// From packages/zql/src/ivm/change.ts
export type Change = AddChange | RemoveChange | ChildChange | EditChange;

// A node (row + relationships) being added
export type AddChange = {
  type: 'add';
  node: Node;
};

// A node being removed
export type RemoveChange = {
  type: 'remove';
  node: Node;
};

// A child relationship changing
export type ChildChange = {
  type: 'child';
  node: Node;
  child: {
    relationshipName: string;
    change: Change;
  };
};

// A row's data changing (PK may change)
export type EditChange = {
  type: 'edit';
  node: Node;
  oldNode: Node;
};

// From packages/zql/src/ivm/operator.ts
export interface Input {
  // Fetch initial data
  fetch(req: FetchRequest): Stream<Node | 'yield'>;
  
  // Receive and transform change
  push(change: Change, pusher: InputBase): Stream<'yield'>;
  
  // Set where to send output
  setOutput(output: Output): void;
}

export interface Output {
  // Push transformed change downstream
  push(change: Change, pusher: InputBase): Stream<'yield'>;
}

// Simplified filter logic
class FilterOperator implements Input, Output {
  push(change: Change): Stream<'yield'> {
    switch (change.type) {
      case 'add':
        // Check if new row matches filter
        if (this.condition(change.node.row)) {
          yield* this.output.push(change);
        }
        break;
        
      case 'remove':
        // If row was in view, remove it
        if (this.storage.has(change.node)) {
          yield* this.output.push(change);
        }
        break;
        
      case 'edit':
        const wasMatch = this.condition(change.oldNode.row);
        const isMatch = this.condition(change.node.row);
        
        if (wasMatch && isMatch) {
          // Still matches, pass through edit
          yield* this.output.push(change);
        } else if (wasMatch && !isMatch) {
          // No longer matches, convert to remove
          yield* this.output.push({type: 'remove', node: change.oldNode});
        } else if (!wasMatch && isMatch) {
          // Now matches, convert to add
          yield* this.output.push({type: 'add', node: change.node});
        }
        break;
    }
  }
}

// Client sends its current version
type PullRequestMessage = {
  pullVersion: number;
  profileID: string;
  clientGroupID: string;
  cookie: NullableVersion;  // Last known server version
};

// Server responds with patch
type PullResponseMessage = {
  cookie: NullableVersion;  // New server version
  lastMutationIDChanges: Record<ClientID, number>;
  patch: Patch[];  // Changes since client's version
};

// Client pushes mutations
type PushMessage = {
  profileID: string;
  clientGroupID: string;
  mutations: Array<CRUDMutation | CustomMutation>;
};

// CRUD mutation example
type CRUDMutation = {
  id: MutationID;
  name: 'zero:crud';
  args: [
    {
      tableName: string;
      op: 'insert' | 'update' | 'delete';
      value: Record<string, unknown>;
    },
  ];
};

type PokeStartMessage = {
  type: 'pokeStart';
  baseCookie: NullableVersion;
  pokeID: string;
};

type PokePartMessage = {
  type: 'pokePart';
  pokeID: string;
  clientsPatch: ClientsPatch[];
};

type PokeEndMessage = {
  type: 'pokeEnd';
  pokeID: string;
  cookie: NullableVersion;
};

// Mutation applies immediately to local IndexedDB
await zero.mutate.updateIssue({
  id: 'issue-123',
  title: 'Updated title',  // Shows immediately in UI
});

// Meanwhile:
// 1. Mutation queued for push
// 2. Pushed to server over WebSocket
// 3. Server validates and executes
// 4. Replication streams back to zero-cache
// 5. zero-cache pokes client
// 6. Client pulls and reconciles

// Example: Two clients update the same issue

// Client A:
await zero.mutate.updateIssue({id: 'issue-1', title: 'New Title A'});

// Client B (simultaneously):
await zero.mutate.updateIssue({id: 'issue-1', title: 'New Title B'});

// Result: Whichever mutation reaches PostgreSQL first wins.
// The other client's optimistic update is replaced by the server's version.

const zero = new Zero({
  server: 'https://your-server.com',
  schema,
  userID,
  // Mutations are queued locally
  // Queries read from IndexedDB
});

// Works offline:
const issues = useQuery(zero.query.issue.where('status', 'open'));

await zero.mutate.createIssue({
  id: nanoid(),
  title: 'Created offline',
  status: 'open',
});

// When online, mutations sync automatically

// Operators yield to avoid blocking
export interface Input {
  fetch(req: FetchRequest): Stream<Node | 'yield'>;
}

// Consumer must handle 'yield' and give control back periodically
for (const item of stream) {
  if (item === 'yield') {
    await sleep(0);  // Let other tasks run
  } else {
    // Process node
  }
}