1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
import type {LogContext} from '@rocicorp/logger';import type {ReadonlyJSONValue} from '../json';import * as db from '../db/mod';import type * as dag from '../dag/mod';import {assertHTTPRequestInfo, HTTPRequestInfo} from '../http-request-info';import {Pusher, PushError} from '../pusher';import {callJSRequest} from './js-request';import {toError} from '../to-error';
export const PUSH_VERSION = 0;
/** * The JSON value used as the body when doing a POST to the [push * endpoint](/server-push). */export type PushRequest = {  profileID: string;  clientID: string;  mutations: Mutation[];  pushVersion: number;  // schema_version can optionally be used to specify to the push endpoint  // version information about the mutators the app is using (e.g., format  // of mutator args).  schemaVersion: string;};
/** * Mutation describes a single mutation done on the client. */export type Mutation = {  readonly id: number;  readonly name: string;  readonly args: ReadonlyJSONValue;  readonly timestamp: number;};
export function convert(lm: db.LocalMeta): Mutation {  return {    id: lm.mutationID,    name: lm.mutatorName,    args: lm.mutatorArgsJSON,    timestamp: lm.timestamp,  };}
export async function push(  requestID: string,  store: dag.Store,  lc: LogContext,  profileID: string,  clientID: string,  pusher: Pusher,  pushURL: string,  auth: string,  schemaVersion: string,): Promise<HTTPRequestInfo | undefined> {  // Find pending commits between the base snapshot and the main head and push  // them to the data layer.  const pending = await store.withRead(async dagRead => {    const mainHeadHash = await dagRead.getHead(db.DEFAULT_HEAD_NAME);    if (!mainHeadHash) {      throw new Error('Internal no main head');    }    return await db.localMutations(mainHeadHash, dagRead);    // Important! Don't hold the lock through an HTTP request!  });  // Commit.pending gave us commits in head-first order; the bindings  // want tail first (in mutation id order).  pending.reverse();
  let httpRequestInfo: HTTPRequestInfo | undefined = undefined;  if (pending.length > 0) {    const pushMutations: Mutation[] = [];    for (const commit of pending) {      if (commit.isLocal()) {        pushMutations.push(convert(commit.meta));      } else {        throw new Error('Internal non local pending commit');      }    }    const pushReq = {      profileID,      clientID,      mutations: pushMutations,      pushVersion: PUSH_VERSION,      schemaVersion,    };    lc.debug?.('Starting push...');    const pushStart = Date.now();    httpRequestInfo = await callPusher(      pusher,      pushURL,      pushReq,      auth,      requestID,    );    lc.debug?.('...Push complete in ', Date.now() - pushStart, 'ms');  }
  return httpRequestInfo;}
async function callPusher(  pusher: Pusher,  url: string,  body: PushRequest,  auth: string,  requestID: string,): Promise<HTTPRequestInfo> {  try {    const res = await callJSRequest(pusher, url, body, auth, requestID);    assertHTTPRequestInfo(res);    return res;  } catch (e) {    throw new PushError(toError(e));  }}