Line data Source code
1 : //
2 : // cloudsync_sqlite.c
3 : // cloudsync
4 : //
5 : // Created by Marco Bambini on 05/12/25.
6 : //
7 :
8 : #include "cloudsync_sqlite.h"
9 : #include "cloudsync_changes_sqlite.h"
10 : #include "../pk.h"
11 : #include "../cloudsync.h"
12 : #include "../block.h"
13 : #include "../database.h"
14 : #include "../dbutils.h"
15 : #include "../sql.h"
16 : #include <inttypes.h>
17 : #include <stddef.h>
18 :
19 : #ifndef CLOUDSYNC_OMIT_NETWORK
20 : #include "../network/network.h"
21 : #endif
22 :
23 : #ifndef SQLITE_CORE
24 : SQLITE_EXTENSION_INIT1
25 : #endif
26 :
27 : #ifndef UNUSED_PARAMETER
28 : #define UNUSED_PARAMETER(X) (void)(X)
29 : #endif
30 :
31 : #ifdef _WIN32
32 : #define APIEXPORT __declspec(dllexport)
33 : #else
34 : #define APIEXPORT
35 : #endif
36 :
37 : typedef struct {
38 : sqlite3_context *context;
39 : int index;
40 : } cloudsync_pk_decode_context;
41 :
42 : typedef struct {
43 : sqlite3_value *table_name;
44 : sqlite3_value **new_values;
45 : sqlite3_value **old_values;
46 : int count;
47 : int capacity;
48 : } cloudsync_update_payload;
49 :
50 : // Reports a failed tracking write from the insert/update/delete triggers: cloudsync's
51 : // own message when it set one (it names the table and column and carries the database
52 : // error), else SQLite's, and the real result code, so a caller can still tell SQLITE_BUSY
53 : // from a constraint violation. The context error is reset on entry to each trigger, so
54 : // the message is never a stale one.
55 4 : static void dbsync_result_trigger_error (sqlite3_context *context, cloudsync_context *data, int rc) {
56 4 : const char *message = cloudsync_errmsg(data);
57 4 : sqlite3_result_error(context, (message && message[0]) ? message : database_errmsg(data), -1);
58 4 : sqlite3_result_error_code(context, (rc > 0) ? rc : SQLITE_ERROR);
59 4 : }
60 :
61 6 : void dbsync_set_error (sqlite3_context *context, const char *format, ...) {
62 : char buffer[2048];
63 :
64 : va_list arg;
65 6 : va_start (arg, format);
66 6 : vsnprintf(buffer, sizeof(buffer), format, arg);
67 6 : va_end (arg);
68 :
69 6 : if (context) sqlite3_result_error(context, buffer, -1);
70 6 : }
71 :
72 : // MARK: - Public -
73 :
74 8 : void dbsync_version (sqlite3_context *context, int argc, sqlite3_value **argv) {
75 : DEBUG_FUNCTION("cloudsync_version");
76 8 : UNUSED_PARAMETER(argc);
77 8 : UNUSED_PARAMETER(argv);
78 8 : sqlite3_result_text(context, CLOUDSYNC_VERSION, -1, SQLITE_STATIC);
79 8 : }
80 :
81 549 : void dbsync_siteid (sqlite3_context *context, int argc, sqlite3_value **argv) {
82 : DEBUG_FUNCTION("cloudsync_siteid");
83 549 : UNUSED_PARAMETER(argc);
84 549 : UNUSED_PARAMETER(argv);
85 :
86 549 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
87 549 : sqlite3_result_blob(context, cloudsync_siteid(data), UUID_LEN, SQLITE_STATIC);
88 549 : }
89 :
90 6 : void dbsync_db_version (sqlite3_context *context, int argc, sqlite3_value **argv) {
91 : DEBUG_FUNCTION("cloudsync_db_version");
92 6 : UNUSED_PARAMETER(argc);
93 6 : UNUSED_PARAMETER(argv);
94 :
95 : // retrieve context
96 6 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
97 :
98 6 : int rc = cloudsync_dbversion_check_uptodate(data);
99 6 : if (rc != SQLITE_OK) {
100 : // When cloudsync_init was never called, data_version_stmt is NULL and
101 : // database_errmsg() falls back to "not an error", producing the
102 : // confusing "Unable to retrieve db_version (not an error)". Detect the
103 : // uninitialized state and return an actionable message instead. The
104 : // extra check only runs on the error branch, so it costs nothing on
105 : // the sync hot path (merge operations keep going through the normal
106 : // path where rc == SQLITE_OK).
107 1 : if (!cloudsync_context_is_initialized(data)) {
108 1 : dbsync_set_error(context,
109 : "cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') "
110 : "to enable sync on a table before calling cloudsync_db_version().");
111 1 : } else {
112 0 : dbsync_set_error(context, "Unable to retrieve db_version (%s).", database_errmsg(data));
113 : }
114 1 : return;
115 : }
116 :
117 5 : sqlite3_result_int64(context, cloudsync_dbversion(data));
118 6 : }
119 :
120 32209 : void dbsync_db_version_next (sqlite3_context *context, int argc, sqlite3_value **argv) {
121 : DEBUG_FUNCTION("cloudsync_db_version_next");
122 :
123 : // retrieve context
124 32209 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
125 :
126 32209 : sqlite3_int64 merging_version = (argc == 1) ? database_value_int(argv[0]) : CLOUDSYNC_VALUE_NOTSET;
127 32209 : sqlite3_int64 value = cloudsync_dbversion_next(data, merging_version);
128 32209 : if (value == -1) {
129 1 : if (!cloudsync_context_is_initialized(data)) {
130 1 : dbsync_set_error(context,
131 : "cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') "
132 : "to enable sync on a table before calling cloudsync_db_version_next().");
133 1 : } else {
134 0 : dbsync_set_error(context, "Unable to retrieve next_db_version (%s).", database_errmsg(data));
135 : }
136 1 : return;
137 : }
138 :
139 32208 : sqlite3_result_int64(context, value);
140 32209 : }
141 :
142 55 : void dbsync_seq (sqlite3_context *context, int argc, sqlite3_value **argv) {
143 : DEBUG_FUNCTION("cloudsync_seq");
144 :
145 : // retrieve context
146 55 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
147 55 : sqlite3_result_int(context, cloudsync_bumpseq(data));
148 55 : }
149 :
150 5 : void dbsync_uuid (sqlite3_context *context, int argc, sqlite3_value **argv) {
151 : DEBUG_FUNCTION("cloudsync_uuid");
152 :
153 : char value[UUID_STR_MAXLEN];
154 5 : char *uuid = cloudsync_uuid_v7_string(value, true);
155 5 : sqlite3_result_text(context, uuid, -1, SQLITE_TRANSIENT);
156 5 : }
157 :
158 : // cloudsync_uuid_text(blob, [dash_format]) -> canonical UUID string
159 4 : void dbsync_uuid_text (sqlite3_context *context, int argc, sqlite3_value **argv) {
160 : DEBUG_FUNCTION("cloudsync_uuid_text");
161 :
162 4 : if (sqlite3_value_type(argv[0]) == SQLITE_NULL) { sqlite3_result_null(context); return; }
163 4 : if (sqlite3_value_type(argv[0]) != SQLITE_BLOB || sqlite3_value_bytes(argv[0]) != UUID_LEN) {
164 0 : sqlite3_result_error(context, "cloudsync_uuid_text: expected a 16-byte BLOB.", -1);
165 0 : return;
166 : }
167 4 : bool dash_format = (argc > 1) ? (sqlite3_value_int(argv[1]) != 0) : true;
168 : char value[UUID_STR_MAXLEN];
169 4 : cloudsync_uuid_v7_stringify((uint8_t *)sqlite3_value_blob(argv[0]), value, dash_format);
170 4 : sqlite3_result_text(context, value, -1, SQLITE_TRANSIENT);
171 4 : }
172 :
173 : // cloudsync_uuid_blob(text) -> 16-byte UUID blob (accepts dashed/undashed)
174 3 : void dbsync_uuid_blob (sqlite3_context *context, int argc, sqlite3_value **argv) {
175 : DEBUG_FUNCTION("cloudsync_uuid_blob");
176 :
177 3 : if (sqlite3_value_type(argv[0]) == SQLITE_NULL) { sqlite3_result_null(context); return; }
178 3 : const char *str = (const char *)sqlite3_value_text(argv[0]);
179 3 : int len = sqlite3_value_bytes(argv[0]);
180 : uint8_t uuid[UUID_LEN];
181 3 : if (!str || cloudsync_uuid_v7_parse(str, len, uuid) != 0) {
182 0 : sqlite3_result_error(context, "cloudsync_uuid_blob: malformed UUID string.", -1);
183 0 : return;
184 : }
185 3 : sqlite3_result_blob(context, uuid, UUID_LEN, SQLITE_TRANSIENT);
186 3 : }
187 :
188 : // MARK: -
189 :
190 25 : void dbsync_set (sqlite3_context *context, int argc, sqlite3_value **argv) {
191 : DEBUG_FUNCTION("cloudsync_set");
192 :
193 : // sanity check parameters
194 25 : const char *key = (const char *)database_value_text(argv[0]);
195 25 : const char *value = (const char *)database_value_text(argv[1]);
196 :
197 : // silently fails
198 25 : if (key == NULL) return;
199 :
200 25 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
201 25 : dbutils_settings_set_key_value(data, key, value);
202 25 : }
203 :
204 336 : void dbsync_set_column (sqlite3_context *context, int argc, sqlite3_value **argv) {
205 : DEBUG_FUNCTION("cloudsync_set_column");
206 :
207 336 : const char *tbl = (const char *)database_value_text(argv[0]);
208 336 : const char *col = (const char *)database_value_text(argv[1]);
209 336 : const char *key = (const char *)database_value_text(argv[2]);
210 336 : const char *value = (const char *)database_value_text(argv[3]);
211 :
212 336 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
213 :
214 : // Handle block column setup: cloudsync_set_column('tbl', 'col', 'algo', 'block')
215 336 : if (key && value && strcmp(key, "algo") == 0 && strcmp(value, "block") == 0) {
216 331 : int rc = cloudsync_setup_block_column(data, tbl, col, NULL, true);
217 331 : if (rc != DBRES_OK) {
218 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
219 0 : }
220 331 : return;
221 : }
222 :
223 : // Handle delimiter setting: cloudsync_set_column('tbl', 'col', 'delimiter', '\n\n')
224 5 : if (key && strcmp(key, "delimiter") == 0) {
225 3 : cloudsync_table_context *table = table_lookup(data, tbl);
226 3 : if (table) {
227 3 : int col_idx = table_col_index(table, col);
228 3 : if (col_idx >= 0 && table_col_algo(table, col_idx) == col_algo_block) {
229 3 : table_set_col_delimiter(table, col_idx, value);
230 3 : }
231 3 : }
232 3 : }
233 :
234 5 : dbutils_table_settings_set_key_value(data, tbl, col, key, value);
235 336 : }
236 :
237 2 : void dbsync_set_table (sqlite3_context *context, int argc, sqlite3_value **argv) {
238 : DEBUG_FUNCTION("cloudsync_set_table");
239 :
240 2 : const char *tbl = (const char *)database_value_text(argv[0]);
241 2 : const char *key = (const char *)database_value_text(argv[1]);
242 2 : const char *value = (const char *)database_value_text(argv[2]);
243 :
244 2 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
245 2 : dbutils_table_settings_set_key_value(data, tbl, "*", key, value);
246 2 : }
247 :
248 2 : void dbsync_set_schema (sqlite3_context *context, int argc, sqlite3_value **argv) {
249 : DEBUG_FUNCTION("dbsync_set_schema");
250 :
251 2 : const char *schema = (const char *)database_value_text(argv[0]);
252 2 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
253 2 : cloudsync_set_schema(data, schema);
254 2 : }
255 :
256 2 : void dbsync_schema (sqlite3_context *context, int argc, sqlite3_value **argv) {
257 : DEBUG_FUNCTION("dbsync_schema");
258 :
259 2 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
260 2 : const char *schema = cloudsync_schema(data);
261 2 : (schema) ? sqlite3_result_text(context, schema, -1, NULL) : sqlite3_result_null(context);
262 2 : }
263 :
264 2 : void dbsync_table_schema (sqlite3_context *context, int argc, sqlite3_value **argv) {
265 : DEBUG_FUNCTION("dbsync_table_schema");
266 :
267 2 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
268 2 : const char *table_name = (const char *)database_value_text(argv[0]);
269 2 : const char *schema = cloudsync_table_schema(data, table_name);
270 2 : (schema) ? sqlite3_result_text(context, schema, -1, NULL) : sqlite3_result_null(context);
271 2 : }
272 :
273 21167 : void dbsync_is_sync (sqlite3_context *context, int argc, sqlite3_value **argv) {
274 : DEBUG_FUNCTION("cloudsync_is_sync");
275 :
276 21167 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
277 21167 : if (cloudsync_insync(data)) {
278 14384 : sqlite3_result_int(context, 1);
279 14384 : return;
280 : }
281 :
282 6783 : const char *table_name = (const char *)database_value_text(argv[0]);
283 6783 : cloudsync_table_context *table = table_lookup(data, table_name);
284 6783 : sqlite3_result_int(context, (table) ? (table_enabled(table) == 0) : 0);
285 21167 : }
286 :
287 155541 : void dbsync_col_value (sqlite3_context *context, int argc, sqlite3_value **argv) {
288 : // DEBUG_FUNCTION("cloudsync_col_value");
289 :
290 : // argv[0] -> table name
291 : // argv[1] -> column name
292 : // argv[2] -> encoded pk
293 :
294 : // retrieve column name
295 155541 : const char *col_name = (const char *)database_value_text(argv[1]);
296 155541 : if (!col_name) {
297 0 : dbsync_set_error(context, "Column name cannot be NULL");
298 0 : return;
299 : }
300 :
301 : // check for special tombstone value
302 155541 : if (strcmp(col_name, CLOUDSYNC_TOMBSTONE_VALUE) == 0) {
303 4789 : sqlite3_result_null(context);
304 4789 : return;
305 : }
306 :
307 : // lookup table
308 150752 : const char *table_name = (const char *)database_value_text(argv[0]);
309 150752 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
310 150752 : cloudsync_table_context *table = table_lookup(data, table_name);
311 150752 : if (!table) {
312 0 : dbsync_set_error(context, "Unable to retrieve table name %s in clousdsync_colvalue.", table_name);
313 0 : return;
314 : }
315 :
316 : // Block column: if col_name contains \x1F, read from blocks table
317 150752 : if (block_is_block_colname(col_name) && table_has_block_cols(table)) {
318 12235 : dbvm_t *bvm = table_block_value_read_stmt(table);
319 12235 : if (!bvm) {
320 0 : sqlite3_result_null(context);
321 0 : return;
322 : }
323 12235 : int rc = databasevm_bind_blob(bvm, 1, database_value_blob(argv[2]), database_value_bytes(argv[2]));
324 12235 : if (rc != DBRES_OK) { databasevm_reset(bvm); sqlite3_result_error(context, database_errmsg(data), -1); return; }
325 12235 : rc = databasevm_bind_text(bvm, 2, col_name, -1);
326 12235 : if (rc != DBRES_OK) { databasevm_reset(bvm); sqlite3_result_error(context, database_errmsg(data), -1); return; }
327 :
328 12235 : rc = databasevm_step(bvm);
329 12235 : if (rc == SQLITE_ROW) {
330 10474 : sqlite3_result_value(context, database_column_value(bvm, 0));
331 12235 : } else if (rc == SQLITE_DONE) {
332 1761 : sqlite3_result_null(context);
333 1761 : } else {
334 0 : sqlite3_result_error(context, database_errmsg(data), -1);
335 : }
336 12235 : databasevm_reset(bvm);
337 12235 : return;
338 : }
339 :
340 : // extract the right col_value vm associated to the column name
341 138517 : sqlite3_stmt *vm = table_column_lookup(table, col_name, false, NULL);
342 138517 : if (!vm) {
343 0 : sqlite3_result_error(context, "Unable to retrieve column value precompiled statement in clousdsync_colvalue.", -1);
344 0 : return;
345 : }
346 :
347 : // bind primary key values
348 138517 : int rc = pk_decode_prikey((char *)database_value_blob(argv[2]), (size_t)database_value_bytes(argv[2]), pk_decode_bind_callback, (void *)vm);
349 138517 : if (rc < 0) goto cleanup;
350 :
351 : // execute vm
352 138517 : rc = databasevm_step(vm);
353 277034 : if (rc == SQLITE_DONE) {
354 0 : rc = SQLITE_OK;
355 0 : sqlite3_result_text(context, CLOUDSYNC_RLS_RESTRICTED_VALUE, -1, SQLITE_STATIC);
356 138517 : } else if (rc == SQLITE_ROW) {
357 : // store value result
358 138517 : rc = SQLITE_OK;
359 138517 : sqlite3_result_value(context, database_column_value(vm, 0));
360 138517 : }
361 :
362 : cleanup:
363 138517 : if (rc != SQLITE_OK) {
364 0 : sqlite3_result_error(context, database_errmsg(data), -1);
365 0 : }
366 138517 : databasevm_reset(vm);
367 155541 : }
368 :
369 10700 : void dbsync_pk_encode (sqlite3_context *context, int argc, sqlite3_value **argv) {
370 10700 : size_t bsize = 0;
371 10700 : char *buffer = pk_encode_prikey((dbvalue_t **)argv, argc, NULL, &bsize);
372 10700 : if (!buffer || buffer == PRIKEY_NULL_CONSTRAINT_ERROR) {
373 1 : sqlite3_result_null(context);
374 1 : return;
375 : }
376 10699 : sqlite3_result_blob(context, (const void *)buffer, (int)bsize, SQLITE_TRANSIENT);
377 10699 : cloudsync_memory_free(buffer);
378 10700 : }
379 :
380 38 : int dbsync_pk_decode_set_result_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
381 38 : cloudsync_pk_decode_context *decode_context = (cloudsync_pk_decode_context *)xdata;
382 : // decode_context->index is 1 based
383 : // index is 0 based
384 38 : if (decode_context->index != index+1) return SQLITE_OK;
385 :
386 26 : int rc = 0;
387 26 : sqlite3_context *context = decode_context->context;
388 26 : switch (type) {
389 : case SQLITE_INTEGER:
390 3 : sqlite3_result_int64(context, ival);
391 3 : break;
392 :
393 : case SQLITE_FLOAT:
394 4 : sqlite3_result_double(context, dval);
395 4 : break;
396 :
397 : case SQLITE_NULL:
398 0 : sqlite3_result_null(context);
399 0 : break;
400 :
401 : case SQLITE_TEXT:
402 18 : sqlite3_result_text(context, pval, (int)ival, SQLITE_TRANSIENT);
403 18 : break;
404 :
405 : case SQLITE_BLOB:
406 1 : sqlite3_result_blob(context, pval, (int)ival, SQLITE_TRANSIENT);
407 1 : break;
408 : }
409 :
410 26 : return rc;
411 38 : }
412 :
413 :
414 26 : void dbsync_pk_decode (sqlite3_context *context, int argc, sqlite3_value **argv) {
415 26 : const char *pk = (const char *)database_value_blob(argv[0]);
416 26 : int pk_len = database_value_bytes(argv[0]);
417 26 : int i = (int)database_value_int(argv[1]);
418 :
419 26 : cloudsync_pk_decode_context xdata = {.context = context, .index = i};
420 26 : pk_decode_prikey((char *)pk, (size_t)pk_len, dbsync_pk_decode_set_result_callback, &xdata);
421 26 : }
422 :
423 : // MARK: -
424 :
425 6013 : void dbsync_insert (sqlite3_context *context, int argc, sqlite3_value **argv) {
426 : DEBUG_FUNCTION("cloudsync_insert %s", database_value_text(argv[0]));
427 : // debug_values(argc-1, &argv[1]);
428 :
429 : // argv[0] is table name
430 : // argv[1]..[N] is primary key(s)
431 :
432 : // table_cloudsync
433 : // pk -> encode(argc-1, &argv[1])
434 : // col_name -> name
435 : // col_version -> 0/1 +1
436 : // db_version -> check
437 : // site_id 0
438 : // seq -> sqlite_master
439 :
440 : // retrieve context
441 6013 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
442 6013 : cloudsync_reset_error(data);
443 :
444 : // lookup table
445 6013 : const char *table_name = (const char *)database_value_text(argv[0]);
446 6013 : cloudsync_table_context *table = table_lookup(data, table_name);
447 6013 : if (!table) {
448 0 : dbsync_set_error(context, "Unable to retrieve table name %s in cloudsync_insert.", table_name);
449 0 : return;
450 : }
451 :
452 : // encode the primary key values into a buffer
453 : char buffer[1024];
454 6013 : size_t pklen = sizeof(buffer);
455 6013 : char *pk = pk_encode_prikey((dbvalue_t **)&argv[1], table_count_pks(table), buffer, &pklen);
456 6013 : if (!pk) {
457 0 : sqlite3_result_error(context, "Not enough memory to encode the primary key(s).", -1);
458 0 : return;
459 : }
460 6013 : if (pk == PRIKEY_NULL_CONSTRAINT_ERROR) {
461 1 : dbsync_set_error(context, "Insert aborted because primary key in table %s contains NULL values.", table_name);
462 1 : return;
463 : }
464 :
465 : // compute the next database version for tracking changes
466 6012 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
467 :
468 : // check if a row with the same primary key already exists
469 : // if so, this means the row might have been previously deleted (sentinel)
470 6012 : bool pk_exists = table_pk_exists(table, pk, pklen);
471 6012 : int rc = SQLITE_OK;
472 :
473 6012 : if (table_count_cols(table) == 0) {
474 : // if there are no columns other than primary keys, insert a sentinel record
475 97 : rc = local_mark_insert_sentinel_meta(table, pk, pklen, db_version, cloudsync_bumpseq(data));
476 97 : if (rc != SQLITE_OK) goto cleanup;
477 6012 : } else if (pk_exists){
478 : // if a row with the same primary key already exists, update the sentinel record
479 290 : rc = local_update_sentinel(table, pk, pklen, db_version, cloudsync_bumpseq(data));
480 290 : if (rc != SQLITE_OK) goto cleanup;
481 290 : }
482 :
483 : // process each non-primary key column for insert or update
484 19923 : for (int i=0; i<table_count_cols(table); ++i) {
485 13914 : if (table_col_algo(table, i) == col_algo_block) {
486 702 : rc = local_block_insert(data, table, pk, pklen, i, db_version);
487 702 : if (rc != SQLITE_OK) goto cleanup;
488 699 : } else {
489 : // Regular column: mark as inserted or updated in the metadata
490 13212 : rc = local_mark_insert_or_update_meta(table, pk, pklen, table_colname(table, i), db_version, cloudsync_bumpseq(data));
491 13212 : if (rc != SQLITE_OK) goto cleanup;
492 : }
493 19920 : }
494 :
495 : cleanup:
496 6012 : if (rc != SQLITE_OK) dbsync_result_trigger_error(context, data, rc);
497 : // free memory if the primary key was dynamically allocated
498 6012 : if (pk != buffer) cloudsync_memory_free(pk);
499 6013 : }
500 :
501 202 : void dbsync_delete (sqlite3_context *context, int argc, sqlite3_value **argv) {
502 : DEBUG_FUNCTION("cloudsync_delete %s", database_value_text(argv[0]));
503 : // debug_values(argc-1, &argv[1]);
504 :
505 : // retrieve context
506 202 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
507 202 : cloudsync_reset_error(data);
508 :
509 : // lookup table
510 202 : const char *table_name = (const char *)database_value_text(argv[0]);
511 202 : cloudsync_table_context *table = table_lookup(data, table_name);
512 202 : if (!table) {
513 0 : dbsync_set_error(context, "Unable to retrieve table name %s in cloudsync_delete.", table_name);
514 0 : return;
515 : }
516 :
517 : // compute the next database version for tracking changes
518 202 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
519 202 : int rc = SQLITE_OK;
520 :
521 : // encode the primary key values into a buffer
522 : char buffer[1024];
523 202 : size_t pklen = sizeof(buffer);
524 202 : char *pk = pk_encode_prikey((dbvalue_t **)&argv[1], table_count_pks(table), buffer, &pklen);
525 202 : if (!pk) {
526 0 : sqlite3_result_error(context, "Not enough memory to encode the primary key(s).", -1);
527 0 : return;
528 : }
529 :
530 202 : if (pk == PRIKEY_NULL_CONSTRAINT_ERROR) {
531 0 : dbsync_set_error(context, "Delete aborted because primary key in table %s contains NULL values.", table_name);
532 0 : return;
533 : }
534 :
535 : // mark the row as deleted by inserting a delete sentinel into the metadata
536 202 : rc = local_mark_delete_meta(table, pk, pklen, db_version, cloudsync_bumpseq(data));
537 202 : if (rc != SQLITE_OK) goto cleanup;
538 :
539 : // remove any metadata related to the old rows associated with this primary key
540 202 : rc = local_drop_meta(table, pk, pklen);
541 202 : if (rc != SQLITE_OK) goto cleanup;
542 :
543 : cleanup:
544 202 : if (rc != SQLITE_OK) dbsync_result_trigger_error(context, data, rc);
545 : // free memory if the primary key was dynamically allocated
546 202 : if (pk != buffer) cloudsync_memory_free(pk);
547 202 : }
548 :
549 : // MARK: -
550 :
551 614 : void dbsync_update_payload_free (cloudsync_update_payload *payload) {
552 3317 : for (int i=0; i<payload->count; i++) {
553 2703 : database_value_free(payload->new_values[i]);
554 2703 : database_value_free(payload->old_values[i]);
555 2703 : }
556 614 : cloudsync_memory_free(payload->new_values);
557 614 : cloudsync_memory_free(payload->old_values);
558 614 : database_value_free(payload->table_name);
559 614 : payload->new_values = NULL;
560 614 : payload->old_values = NULL;
561 614 : payload->table_name = NULL;
562 614 : payload->count = 0;
563 614 : payload->capacity = 0;
564 614 : }
565 :
566 2703 : int dbsync_update_payload_append (cloudsync_update_payload *payload, sqlite3_value *v1, sqlite3_value *v2, sqlite3_value *v3) {
567 2703 : if (payload->count >= payload->capacity) {
568 617 : int newcap = payload->capacity ? payload->capacity * 2 : 128;
569 :
570 617 : sqlite3_value **new_values_2 = (sqlite3_value **)cloudsync_memory_realloc(payload->new_values, newcap * sizeof(*new_values_2));
571 617 : if (!new_values_2) return SQLITE_NOMEM;
572 :
573 617 : sqlite3_value **old_values_2 = (sqlite3_value **)cloudsync_memory_realloc(payload->old_values, newcap * sizeof(*old_values_2));
574 617 : if (!old_values_2) {
575 : // new_values_2 succeeded but old_values failed; keep new_values_2 pointer
576 : // (it's still valid, just larger) but don't update capacity
577 0 : payload->new_values = new_values_2;
578 0 : return SQLITE_NOMEM;
579 : }
580 :
581 617 : payload->new_values = new_values_2;
582 617 : payload->old_values = old_values_2;
583 617 : payload->capacity = newcap;
584 617 : }
585 :
586 2703 : int index = payload->count;
587 2703 : if (payload->table_name == NULL) payload->table_name = database_value_dup(v1);
588 2089 : else if (dbutils_value_compare(payload->table_name, v1) != 0) return SQLITE_NOMEM;
589 :
590 2703 : payload->new_values[index] = database_value_dup(v2);
591 2703 : payload->old_values[index] = database_value_dup(v3);
592 :
593 : // sanity check memory allocations before committing count
594 2703 : bool v1_can_be_null = (database_value_type(v1) == SQLITE_NULL);
595 2703 : bool v2_can_be_null = (database_value_type(v2) == SQLITE_NULL);
596 2703 : bool v3_can_be_null = (database_value_type(v3) == SQLITE_NULL);
597 :
598 2703 : bool oom = false;
599 2703 : if ((payload->table_name == NULL) && (!v1_can_be_null)) oom = true;
600 2703 : if ((payload->new_values[index] == NULL) && (!v2_can_be_null)) oom = true;
601 2703 : if ((payload->old_values[index] == NULL) && (!v3_can_be_null)) oom = true;
602 :
603 2703 : if (oom) {
604 : // clean up partial allocations at this index to prevent leaks
605 0 : if (payload->new_values[index]) { database_value_free(payload->new_values[index]); payload->new_values[index] = NULL; }
606 0 : if (payload->old_values[index]) { database_value_free(payload->old_values[index]); payload->old_values[index] = NULL; }
607 0 : return SQLITE_NOMEM;
608 : }
609 :
610 2703 : payload->count++;
611 2703 : return SQLITE_OK;
612 2703 : }
613 :
614 2703 : void dbsync_update_step (sqlite3_context *context, int argc, sqlite3_value **argv) {
615 : // argv[0] => table_name
616 : // argv[1] => new_column_value
617 : // argv[2] => old_column_value
618 :
619 : // allocate/get the update payload
620 2703 : cloudsync_update_payload *payload = (cloudsync_update_payload *)sqlite3_aggregate_context(context, sizeof(cloudsync_update_payload));
621 2703 : if (!payload) {sqlite3_result_error_nomem(context); return;}
622 :
623 2703 : if (dbsync_update_payload_append(payload, argv[0], argv[1], argv[2]) != SQLITE_OK) {
624 0 : sqlite3_result_error_nomem(context);
625 0 : }
626 2703 : }
627 :
628 614 : void dbsync_update_final (sqlite3_context *context) {
629 614 : cloudsync_update_payload *payload = (cloudsync_update_payload *)sqlite3_aggregate_context(context, sizeof(cloudsync_update_payload));
630 614 : if (!payload || payload->count == 0) return;
631 :
632 : // retrieve context
633 614 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
634 614 : cloudsync_reset_error(data);
635 :
636 : // lookup table
637 614 : const char *table_name = (const char *)database_value_text(payload->table_name);
638 614 : cloudsync_table_context *table = table_lookup(data, table_name);
639 614 : if (!table) {
640 0 : dbsync_set_error(context, "Unable to retrieve table name %s in cloudsync_update.", table_name);
641 0 : dbsync_update_payload_free(payload);
642 0 : return;
643 : }
644 :
645 : // compute the next database version for tracking changes
646 614 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
647 614 : int rc = SQLITE_OK;
648 :
649 : // Check if the primary key(s) have changed
650 614 : bool prikey_changed = false;
651 1510 : for (int i=0; i<table_count_pks(table); ++i) {
652 924 : if (dbutils_value_compare(payload->old_values[i], payload->new_values[i]) != 0) {
653 28 : prikey_changed = true;
654 28 : break;
655 : }
656 896 : }
657 :
658 : // encode the NEW primary key values into a buffer (used later for indexing)
659 : char buffer[1024];
660 : char buffer2[1024];
661 614 : size_t pklen = sizeof(buffer);
662 614 : size_t oldpklen = sizeof(buffer2);
663 614 : char *oldpk = NULL;
664 :
665 614 : char *pk = pk_encode_prikey((dbvalue_t **)payload->new_values, table_count_pks(table), buffer, &pklen);
666 614 : if (!pk) {
667 0 : sqlite3_result_error(context, "Not enough memory to encode the primary key(s).", -1);
668 0 : dbsync_update_payload_free(payload);
669 0 : return;
670 : }
671 614 : if (pk == PRIKEY_NULL_CONSTRAINT_ERROR) {
672 0 : dbsync_set_error(context, "Update aborted because primary key in table %s contains NULL values.", table_name);
673 0 : dbsync_update_payload_free(payload);
674 0 : return;
675 : }
676 :
677 614 : if (prikey_changed) {
678 : // if the primary key has changed, we need to handle the row differently:
679 : // 1. mark the old row (OLD primary key) as deleted
680 : // 2. create a new row (NEW primary key)
681 :
682 : // encode the OLD primary key into a buffer
683 28 : oldpk = pk_encode_prikey((dbvalue_t **)payload->old_values, table_count_pks(table), buffer2, &oldpklen);
684 28 : if (!oldpk) {
685 : // no check here about PRIKEY_NULL_CONSTRAINT_ERROR because by design oldpk cannot contain NULL values
686 0 : if (pk != buffer) cloudsync_memory_free(pk);
687 0 : sqlite3_result_error(context, "Not enough memory to encode the primary key(s).", -1);
688 0 : dbsync_update_payload_free(payload);
689 0 : return;
690 : }
691 :
692 : // mark the rows with the old primary key as deleted in the metadata (old row handling)
693 28 : rc = local_mark_delete_meta(table, oldpk, oldpklen, db_version, cloudsync_bumpseq(data));
694 28 : if (rc != SQLITE_OK) goto cleanup;
695 :
696 : // move non-sentinel metadata entries from OLD primary key to NEW primary key
697 : // handles the case where some metadata is retained across primary key change
698 : // see https://github.com/sqliteai/sqlite-sync/blob/main/docs/PriKey.md for more details
699 28 : rc = local_update_move_meta(table, pk, pklen, oldpk, oldpklen, db_version);
700 28 : if (rc != SQLITE_OK) goto cleanup;
701 :
702 : // mark a new sentinel row with the new primary key in the metadata
703 28 : rc = local_mark_insert_sentinel_meta(table, pk, pklen, db_version, cloudsync_bumpseq(data));
704 28 : if (rc != SQLITE_OK) goto cleanup;
705 :
706 : // free memory if the OLD primary key was dynamically allocated
707 28 : if (oldpk != buffer2) cloudsync_memory_free(oldpk);
708 28 : oldpk = NULL;
709 28 : }
710 :
711 : // compare NEW and OLD values (excluding primary keys) to handle column updates
712 2365 : for (int i=0; i<table_count_cols(table); i++) {
713 1752 : int col_index = table_count_pks(table) + i; // Regular columns start after primary keys
714 :
715 1752 : if (dbutils_value_compare(payload->old_values[col_index], payload->new_values[col_index]) != 0) {
716 938 : if (table_col_algo(table, i) == col_algo_block) {
717 510 : rc = local_block_update(data, table, pk, pklen, i,
718 255 : (const char *)database_value_text(payload->new_values[col_index]), db_version, false);
719 255 : if (rc != SQLITE_OK) goto cleanup;
720 254 : } else {
721 : // Regular column: mark as updated in the metadata (columns are in cid order)
722 683 : rc = local_mark_insert_or_update_meta(table, pk, pklen, table_colname(table, i), db_version, cloudsync_bumpseq(data));
723 683 : if (rc != SQLITE_OK) goto cleanup;
724 : }
725 937 : }
726 2364 : }
727 :
728 : cleanup:
729 614 : if (rc != SQLITE_OK) dbsync_result_trigger_error(context, data, rc);
730 614 : if (pk != buffer) cloudsync_memory_free(pk);
731 614 : if (oldpk && (oldpk != buffer2)) cloudsync_memory_free(oldpk);
732 :
733 614 : dbsync_update_payload_free(payload);
734 614 : }
735 :
736 : // MARK: -
737 :
738 4 : void dbsync_cleanup (sqlite3_context *context, int argc, sqlite3_value **argv) {
739 : DEBUG_FUNCTION("cloudsync_cleanup");
740 :
741 4 : const char *table = (const char *)database_value_text(argv[0]);
742 4 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
743 :
744 4 : int rc = cloudsync_cleanup(data, table);
745 4 : if (rc != DBRES_OK) {
746 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
747 0 : sqlite3_result_error_code(context, rc);
748 0 : }
749 4 : }
750 :
751 10 : void dbsync_enable_disable (sqlite3_context *context, const char *table_name, bool value) {
752 : DEBUG_FUNCTION("cloudsync_enable_disable");
753 :
754 10 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
755 10 : cloudsync_table_context *table = table_lookup(data, table_name);
756 10 : if (!table) return;
757 :
758 10 : table_set_enabled(table, value);
759 10 : }
760 :
761 5 : void dbsync_enable (sqlite3_context *context, int argc, sqlite3_value **argv) {
762 : DEBUG_FUNCTION("cloudsync_enable");
763 :
764 5 : const char *table = (const char *)database_value_text(argv[0]);
765 5 : dbsync_enable_disable(context, table, true);
766 5 : }
767 :
768 5 : void dbsync_disable (sqlite3_context *context, int argc, sqlite3_value **argv) {
769 : DEBUG_FUNCTION("cloudsync_disable");
770 :
771 5 : const char *table = (const char *)database_value_text(argv[0]);
772 5 : dbsync_enable_disable(context, table, false);
773 5 : }
774 :
775 2858 : void dbsync_is_enabled (sqlite3_context *context, int argc, sqlite3_value **argv) {
776 : DEBUG_FUNCTION("cloudsync_is_enabled");
777 :
778 2858 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
779 2858 : const char *table_name = (const char *)database_value_text(argv[0]);
780 2858 : cloudsync_table_context *table = table_lookup(data, table_name);
781 :
782 2858 : int result = (table && table_enabled(table)) ? 1 : 0;
783 2858 : sqlite3_result_int(context, result);
784 2858 : }
785 :
786 772 : void dbsync_terminate (sqlite3_context *context, int argc, sqlite3_value **argv) {
787 : DEBUG_FUNCTION("cloudsync_terminate");
788 :
789 772 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
790 772 : int rc = cloudsync_terminate(data);
791 772 : sqlite3_result_int(context, rc);
792 772 : }
793 :
794 : // MARK: -
795 :
796 801 : void dbsync_init (sqlite3_context *context, const char *table, const char *algo, CLOUDSYNC_INIT_FLAG init_flags) {
797 801 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
798 :
799 801 : int rc = database_begin_savepoint(data, "cloudsync_init");
800 801 : if (rc != SQLITE_OK) {
801 0 : dbsync_set_error(context, "Unable to create cloudsync_init savepoint. %s", database_errmsg(data));
802 0 : sqlite3_result_error_code(context, rc);
803 0 : return;
804 : }
805 :
806 801 : rc = cloudsync_init_table(data, table, algo, init_flags);
807 801 : if (rc == SQLITE_OK) {
808 796 : rc = database_commit_savepoint(data, "cloudsync_init");
809 796 : if (rc != SQLITE_OK) {
810 0 : dbsync_set_error(context, "Unable to release cloudsync_init savepoint. %s", database_errmsg(data));
811 0 : sqlite3_result_error_code(context, rc);
812 0 : }
813 796 : } else {
814 : // in case of error, rollback transaction
815 5 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
816 5 : sqlite3_result_error_code(context, rc);
817 5 : database_rollback_savepoint(data, "cloudsync_init");
818 5 : return;
819 : }
820 :
821 796 : cloudsync_update_schema_hash(data);
822 :
823 : // returns site_id as TEXT
824 : char buffer[UUID_STR_MAXLEN];
825 796 : cloudsync_uuid_v7_stringify(cloudsync_siteid(data), buffer, false);
826 796 : sqlite3_result_text(context, buffer, -1, SQLITE_TRANSIENT);
827 801 : }
828 :
829 46 : void dbsync_init3 (sqlite3_context *context, int argc, sqlite3_value **argv) {
830 : DEBUG_FUNCTION("cloudsync_init2");
831 :
832 46 : const char *table = (const char *)database_value_text(argv[0]);
833 46 : const char *algo = (const char *)database_value_text(argv[1]);
834 46 : int init_flags = database_value_int(argv[2]);
835 46 : dbsync_init(context, table, algo, init_flags);
836 46 : }
837 :
838 86 : void dbsync_init2 (sqlite3_context *context, int argc, sqlite3_value **argv) {
839 : DEBUG_FUNCTION("cloudsync_init2");
840 :
841 86 : const char *table = (const char *)database_value_text(argv[0]);
842 86 : const char *algo = (const char *)database_value_text(argv[1]);
843 86 : dbsync_init(context, table, algo, CLOUDSYNC_INIT_FLAG_NONE);
844 86 : }
845 :
846 669 : void dbsync_init1 (sqlite3_context *context, int argc, sqlite3_value **argv) {
847 : DEBUG_FUNCTION("cloudsync_init1");
848 :
849 669 : const char *table = (const char *)database_value_text(argv[0]);
850 669 : dbsync_init(context, table, NULL, CLOUDSYNC_INIT_FLAG_NONE);
851 669 : }
852 :
853 : // MARK: -
854 :
855 24 : void dbsync_begin_alter (sqlite3_context *context, int argc, sqlite3_value **argv) {
856 : DEBUG_FUNCTION("dbsync_begin_alter");
857 :
858 24 : const char *table_name = (const char *)database_value_text(argv[0]);
859 24 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
860 :
861 24 : int rc = database_begin_savepoint(data, "cloudsync_alter");
862 24 : if (rc != DBRES_OK) {
863 0 : sqlite3_result_error(context, "Unable to create cloudsync_alter savepoint", -1);
864 0 : sqlite3_result_error_code(context, rc);
865 0 : return;
866 : }
867 :
868 24 : rc = cloudsync_begin_alter(data, table_name);
869 24 : if (rc != DBRES_OK) {
870 1 : database_rollback_savepoint(data, "cloudsync_alter");
871 1 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
872 1 : sqlite3_result_error_code(context, rc);
873 1 : }
874 24 : }
875 :
876 24 : void dbsync_commit_alter (sqlite3_context *context, int argc, sqlite3_value **argv) {
877 : DEBUG_FUNCTION("cloudsync_commit_alter");
878 :
879 : //retrieve table argument
880 24 : const char *table_name = (const char *)database_value_text(argv[0]);
881 :
882 : // retrieve context
883 24 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
884 :
885 24 : int rc = cloudsync_commit_alter(data, table_name);
886 24 : if (rc != DBRES_OK) {
887 1 : database_rollback_savepoint(data, "cloudsync_alter");
888 1 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
889 1 : sqlite3_result_error_code(context, rc);
890 1 : return;
891 : }
892 :
893 23 : rc = database_commit_savepoint(data, "cloudsync_alter");
894 23 : if (rc != DBRES_OK) {
895 0 : sqlite3_result_error(context, database_errmsg(data), -1);
896 0 : sqlite3_result_error_code(context, rc);
897 0 : return;
898 : }
899 :
900 23 : cloudsync_update_schema_hash(data);
901 24 : }
902 :
903 : // MARK: - Payload -
904 :
905 56708 : void dbsync_payload_encode_step (sqlite3_context *context, int argc, sqlite3_value **argv) {
906 : // allocate/get the session context
907 56708 : cloudsync_payload_context *payload = (cloudsync_payload_context *)sqlite3_aggregate_context(context, (int)cloudsync_payload_context_size(NULL));
908 56708 : if (!payload) {
909 0 : sqlite3_result_error(context, "Not enough memory to allocate payload session context", -1);
910 0 : sqlite3_result_error_code(context, SQLITE_NOMEM);
911 0 : return;
912 : }
913 :
914 : // retrieve context
915 56708 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
916 :
917 56708 : int rc = cloudsync_payload_encode_step(payload, data, argc, (dbvalue_t **)argv);
918 56708 : if (rc != SQLITE_OK) {
919 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
920 0 : sqlite3_result_error_code(context, rc);
921 0 : }
922 56708 : }
923 :
924 1810 : void dbsync_payload_encode_final (sqlite3_context *context) {
925 : // get the session context
926 1810 : cloudsync_payload_context *payload = (cloudsync_payload_context *)sqlite3_aggregate_context(context, (int)cloudsync_payload_context_size(NULL));
927 1810 : if (!payload) {
928 0 : sqlite3_result_error(context, "Unable to extract payload session context", -1);
929 0 : sqlite3_result_error_code(context, SQLITE_NOMEM);
930 0 : return;
931 : }
932 :
933 : // retrieve context
934 1810 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
935 :
936 1810 : int rc = cloudsync_payload_encode_final(payload, data);
937 1810 : if (rc != SQLITE_OK) {
938 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
939 0 : sqlite3_result_error_code(context, rc);
940 0 : return;
941 : }
942 :
943 : // result is OK so get BLOB and returns it
944 1810 : int64_t blob_size = 0;
945 1810 : char *blob = cloudsync_payload_blob (payload, &blob_size, NULL);
946 1810 : if (!blob) {
947 8 : sqlite3_result_null(context);
948 8 : } else {
949 1802 : sqlite3_result_blob64(context, blob, blob_size, SQLITE_TRANSIENT);
950 1802 : cloudsync_memory_free(blob);
951 : }
952 :
953 : // from: https://sqlite.org/c3ref/aggregate_context.html
954 : // SQLite automatically frees the memory allocated by sqlite3_aggregate_context() when the aggregate query concludes.
955 1810 : }
956 :
957 1884 : void dbsync_payload_decode (sqlite3_context *context, int argc, sqlite3_value **argv) {
958 : DEBUG_FUNCTION("dbsync_payload_decode");
959 : //debug_values(argc, argv);
960 :
961 : // sanity check payload type
962 1884 : if (database_value_type(argv[0]) != SQLITE_BLOB) {
963 0 : sqlite3_result_error(context, "Error on cloudsync_payload_decode: value must be a BLOB.", -1);
964 0 : sqlite3_result_error_code(context, SQLITE_MISUSE);
965 0 : return;
966 : }
967 :
968 : // sanity check payload size
969 1884 : int blen = database_value_bytes(argv[0]);
970 1884 : size_t header_size = 0;
971 1884 : cloudsync_payload_context_size(&header_size);
972 1884 : if (blen < (int)header_size) {
973 3 : sqlite3_result_error(context, "Error on cloudsync_payload_decode: invalid input size.", -1);
974 3 : sqlite3_result_error_code(context, SQLITE_MISUSE);
975 3 : return;
976 : }
977 :
978 : // obtain payload
979 1881 : const char *payload = (const char *)database_value_blob(argv[0]);
980 :
981 : // apply changes
982 : // The public SQL function applies a single complete payload: advance the
983 : // receive cursor to its last applied (db_version, seq) (legacy behavior, safe
984 : // for a payload that ends on a db_version boundary). The chunked-download
985 : // receive path gates cursor advancement on stream completion via the C-level
986 : // checkpoint argument instead (see cloudsync_payload_apply in cloudsync.h).
987 1881 : int nrows = 0;
988 1881 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
989 1881 : int rc = cloudsync_payload_apply(data, payload, blen, &nrows, CLOUDSYNC_CHECKPOINT_LAST_APPLIED, 0);
990 1881 : if (rc != SQLITE_OK) {
991 272 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
992 272 : sqlite3_result_error_code(context, rc);
993 272 : return;
994 : }
995 :
996 : // returns number of applied rows
997 1609 : sqlite3_result_int(context, nrows);
998 1884 : }
999 :
1000 : typedef struct {
1001 : sqlite3_vtab base;
1002 : sqlite3 *db;
1003 : cloudsync_context *data;
1004 : } cloudsync_payload_chunks_vtab;
1005 :
1006 : typedef struct {
1007 : sqlite3_vtab_cursor base;
1008 : cloudsync_payload_chunks_vtab *vtab;
1009 : sqlite3_stmt *src;
1010 : bool eof;
1011 : bool has_row;
1012 : int chunk_index;
1013 : char *payload;
1014 : int64_t payload_size;
1015 : int64_t rows;
1016 : int64_t dbv_min;
1017 : int64_t dbv_max;
1018 : int64_t watermark;
1019 : bool frag_active;
1020 : int frag_part;
1021 : int frag_count;
1022 : int frag_target;
1023 : int64_t frag_offset;
1024 : int64_t frag_total;
1025 : uint64_t frag_checksum;
1026 : char value_header[16];
1027 : int value_header_len;
1028 : const char *value_data;
1029 : int64_t value_data_len;
1030 : // Positional-cursor outputs: the resume point AFTER the chunk currently held.
1031 : // These live in the per-scan reset region (after eof) so xFilter's bulk memset
1032 : // clears them. next_* is the (db_version, seq, frag_offset) a follow-up call
1033 : // passes back as resume_* to continue exactly where this chunk stopped.
1034 : int64_t next_dbv;
1035 : int64_t next_seq;
1036 : int64_t next_frag_offset;
1037 : bool is_final;
1038 : // Window cap (max_window_bytes): payload bytes emitted so far by this scan, and
1039 : // whether the scan ended because the budget ran out rather than because the
1040 : // window was drained. window_bytes counts across chunks, not within one.
1041 : int64_t max_window_bytes;
1042 : int64_t window_bytes;
1043 : bool window_capped;
1044 : } cloudsync_payload_chunks_cursor;
1045 :
1046 19 : static int payload_chunks_connect(sqlite3 *db, void *aux, int argc, const char *const *argv, sqlite3_vtab **vtab, char **err) {
1047 19 : UNUSED_PARAMETER(argc); UNUSED_PARAMETER(argv); UNUSED_PARAMETER(err);
1048 19 : int rc = sqlite3_declare_vtab(db,
1049 : "CREATE TABLE x(payload BLOB, chunk_index INTEGER, payload_size INTEGER, rows INTEGER, "
1050 : "db_version_min INTEGER, db_version_max INTEGER, watermark_db_version INTEGER, "
1051 : "since_db_version HIDDEN, site_id HIDDEN, until_db_version HIDDEN, exclude_filter_site_id HIDDEN, "
1052 : // Positional-cursor outputs (cols 11..14): the resume point after the
1053 : // emitted chunk, plus a final-chunk flag. A stateless /check passes these
1054 : // back as the resume_* inputs (cols 15..17) to continue the drain without
1055 : // a spool table — O(1) seek per chunk instead of replaying from since.
1056 : "next_db_version INTEGER, next_seq INTEGER, next_frag_offset INTEGER, is_final INTEGER, "
1057 : // window_capped and window_bytes (cols 15..16) are outputs, so they do not shift
1058 : // the hidden columns a table-valued call binds positionally. The two budget
1059 : // inputs are declared last for the same reason: they become arguments 8 and 9,
1060 : // leaving arguments 1..7 as they were.
1061 : //
1062 : // window_bytes/resume_window_bytes carry the budget spent across calls exactly
1063 : // as next_*/resume_* carry the stream position. Without it a caller that fetches
1064 : // one chunk per call -- which is how a stateless /check pages -- restarts the
1065 : // count every time and never reaches the budget at all.
1066 : "window_capped INTEGER, window_bytes INTEGER, "
1067 : "resume_db_version HIDDEN, resume_seq HIDDEN, resume_frag_offset HIDDEN, "
1068 : "max_window_bytes HIDDEN, resume_window_bytes HIDDEN)");
1069 19 : if (rc != SQLITE_OK) return rc;
1070 19 : cloudsync_payload_chunks_vtab *p = sqlite3_malloc64(sizeof(*p));
1071 19 : if (!p) return SQLITE_NOMEM;
1072 19 : memset(p, 0, sizeof(*p));
1073 19 : p->db = db;
1074 19 : p->data = (cloudsync_context *)aux;
1075 19 : *vtab = (sqlite3_vtab *)p;
1076 19 : return SQLITE_OK;
1077 19 : }
1078 :
1079 19 : static int payload_chunks_disconnect(sqlite3_vtab *vtab) {
1080 19 : sqlite3_free(vtab);
1081 19 : return SQLITE_OK;
1082 : }
1083 :
1084 187 : static int payload_chunks_open(sqlite3_vtab *vtab, sqlite3_vtab_cursor **cursor) {
1085 187 : cloudsync_payload_chunks_cursor *c = cloudsync_memory_zeroalloc(sizeof(*c));
1086 187 : if (!c) return SQLITE_NOMEM;
1087 187 : c->vtab = (cloudsync_payload_chunks_vtab *)vtab;
1088 187 : *cursor = (sqlite3_vtab_cursor *)c;
1089 187 : return SQLITE_OK;
1090 187 : }
1091 :
1092 187 : static int payload_chunks_close(sqlite3_vtab_cursor *cursor) {
1093 187 : cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
1094 187 : if (c->src) sqlite3_finalize(c->src);
1095 187 : if (c->payload) cloudsync_memory_free(c->payload);
1096 187 : cloudsync_memory_free(c);
1097 187 : return SQLITE_OK;
1098 : }
1099 :
1100 178 : static int payload_chunks_best_index(sqlite3_vtab *vtab, sqlite3_index_info *idxinfo) {
1101 178 : UNUSED_PARAMETER(vtab);
1102 : // Assign argvIndex in a canonical hidden-column order so xFilter can read argv
1103 : // in a fixed order regardless of how SQLite presents constraints. idxNum bit k
1104 : // is set when handled_cols[k] is bound; xFilter reads argv in this same order.
1105 : // bit0=since_db_version(7) bit1=site_id(8) bit2=until_db_version(9)
1106 : // bit3=exclude_filter_site_id(10) bit4=resume_db_version(17)
1107 : // bit5=resume_seq(18) bit6=resume_frag_offset(19) bit7=max_window_bytes(20)
1108 : // bit8=resume_window_bytes(21)
1109 : static const int handled_cols[] = {7, 8, 9, 10, 17, 18, 19, 20, 21};
1110 178 : int argv_index = 1;
1111 178 : int idxnum = 0;
1112 1780 : for (size_t k = 0; k < sizeof(handled_cols) / sizeof(handled_cols[0]); ++k) {
1113 1602 : int col = handled_cols[k];
1114 6352 : for (int i = 0; i < idxinfo->nConstraint; ++i) {
1115 5753 : struct sqlite3_index_constraint *cn = &idxinfo->aConstraint[i];
1116 5753 : if (!cn->usable || cn->op != SQLITE_INDEX_CONSTRAINT_EQ || cn->iColumn != col) continue;
1117 1003 : idxinfo->aConstraintUsage[i].argvIndex = argv_index++;
1118 1003 : idxinfo->aConstraintUsage[i].omit = 1;
1119 1003 : idxnum |= (1 << k);
1120 1003 : break; // at most one constraint consumed per hidden column
1121 : }
1122 1602 : }
1123 178 : idxinfo->idxNum = idxnum;
1124 178 : idxinfo->estimatedCost = 10.0;
1125 178 : idxinfo->estimatedRows = 10;
1126 178 : return SQLITE_OK;
1127 : }
1128 :
1129 3425 : static int payload_chunks_step_source(cloudsync_payload_chunks_cursor *c) {
1130 3425 : int rc = sqlite3_step(c->src);
1131 3425 : if (rc == SQLITE_ROW) { c->has_row = true; return SQLITE_OK; }
1132 66 : c->has_row = false;
1133 66 : if (rc == SQLITE_DONE) return SQLITE_OK;
1134 : // copy the inner statement's message onto this vtab or SQLite surfaces the
1135 : // error as a bare "SQL logic error"
1136 0 : if (c->vtab->base.zErrMsg) sqlite3_free(c->vtab->base.zErrMsg);
1137 0 : c->vtab->base.zErrMsg = sqlite3_mprintf("%s", sqlite3_errmsg(c->vtab->db));
1138 0 : return rc;
1139 3425 : }
1140 :
1141 61 : static int payload_chunks_plan_fragment(cloudsync_payload_chunks_cursor *c) {
1142 61 : cloudsync_context *data = c->vtab->data;
1143 122 : int target = cloudsync_payload_fragment_data_size(data,
1144 61 : (const char *)sqlite3_column_text(c->src, 0), sqlite3_column_bytes(c->src, 0),
1145 61 : sqlite3_column_blob(c->src, 1), sqlite3_column_bytes(c->src, 1),
1146 61 : (const char *)sqlite3_column_text(c->src, 2), sqlite3_column_bytes(c->src, 2),
1147 61 : sqlite3_column_int64(c->src, 4), sqlite3_column_int64(c->src, 5),
1148 61 : sqlite3_column_blob(c->src, 6), sqlite3_column_bytes(c->src, 6),
1149 61 : sqlite3_column_int64(c->src, 7), sqlite3_column_int64(c->src, 8),
1150 61 : c->frag_total, 0, 1);
1151 61 : if (target <= 0) return SQLITE_TOOBIG;
1152 :
1153 61 : int count = 0;
1154 61 : for (int i = 0; i < CLOUDSYNC_PAYLOAD_FRAGMENT_SIZE_FIXPOINT_ITERATIONS; ++i) {
1155 61 : count = cloudsync_payload_fragment_count(c->frag_total, target);
1156 61 : if (count <= 0) return SQLITE_TOOBIG;
1157 122 : int planned = cloudsync_payload_fragment_data_size(data,
1158 61 : (const char *)sqlite3_column_text(c->src, 0), sqlite3_column_bytes(c->src, 0),
1159 61 : sqlite3_column_blob(c->src, 1), sqlite3_column_bytes(c->src, 1),
1160 61 : (const char *)sqlite3_column_text(c->src, 2), sqlite3_column_bytes(c->src, 2),
1161 61 : sqlite3_column_int64(c->src, 4), sqlite3_column_int64(c->src, 5),
1162 61 : sqlite3_column_blob(c->src, 6), sqlite3_column_bytes(c->src, 6),
1163 61 : sqlite3_column_int64(c->src, 7), sqlite3_column_int64(c->src, 8),
1164 61 : c->frag_total, count - 1, count);
1165 61 : if (planned <= 0) return SQLITE_TOOBIG;
1166 61 : if (planned == target) break;
1167 0 : target = planned;
1168 0 : }
1169 :
1170 61 : c->frag_target = target;
1171 61 : c->frag_count = cloudsync_payload_fragment_count(c->frag_total, target);
1172 61 : if (c->frag_count <= 0) return SQLITE_TOOBIG;
1173 61 : return SQLITE_OK;
1174 61 : }
1175 :
1176 : // Set up fragment state for the current source row (a single value larger than
1177 : // max_chunk_size) so emit_fragment can stream it. start_offset is the byte offset
1178 : // within the encoded value to resume from (0 when first reaching the value;
1179 : // >0 when a positional cursor resumes mid-value). frag_part is derived from the
1180 : // offset so the fragment's part index is consistent whether reached by streaming
1181 : // or by a seek. The plan (frag_target/frag_count) is a deterministic function of
1182 : // the row, so a resumed fragment tiles identically to a streamed one.
1183 61 : static int payload_chunks_begin_fragment(cloudsync_payload_chunks_cursor *c, int64_t start_offset) {
1184 61 : dbvalue_t *col_value = (dbvalue_t *)sqlite3_column_value(c->src, 3);
1185 61 : int type = database_value_type(col_value);
1186 61 : if (type != DBTYPE_TEXT && type != DBTYPE_BLOB) return SQLITE_TOOBIG;
1187 61 : int64_t raw_len = 0;
1188 61 : int header_len = cloudsync_payload_encoded_value_header(col_value, c->value_header, sizeof(c->value_header), &raw_len);
1189 61 : if (header_len <= 0) return SQLITE_ERROR;
1190 61 : c->value_header_len = header_len;
1191 61 : c->value_data = (const char *)database_value_blob(col_value);
1192 61 : c->value_data_len = raw_len;
1193 61 : c->frag_total = header_len + raw_len;
1194 61 : c->frag_offset = start_offset;
1195 61 : int rc = payload_chunks_plan_fragment(c);
1196 61 : if (rc != SQLITE_OK) return rc;
1197 61 : c->frag_part = (c->frag_target > 0) ? (int)(start_offset / c->frag_target) : 0;
1198 61 : c->frag_checksum = cloudsync_payload_encoded_value_checksum(col_value);
1199 61 : c->frag_active = true;
1200 61 : return SQLITE_OK;
1201 61 : }
1202 :
1203 133 : static int payload_chunks_emit_fragment(cloudsync_payload_chunks_cursor *c) {
1204 133 : cloudsync_context *data = c->vtab->data;
1205 133 : if (c->payload) { cloudsync_memory_free(c->payload); c->payload = NULL; }
1206 133 : int64_t remaining = c->frag_total - c->frag_offset;
1207 133 : int frag_len = remaining > c->frag_target ? c->frag_target : (int)remaining;
1208 133 : if (frag_len <= 0) return SQLITE_CORRUPT;
1209 133 : char *frag = cloudsync_memory_alloc((uint64_t)frag_len);
1210 133 : if (!frag) return SQLITE_NOMEM;
1211 133 : int copied = 0;
1212 133 : int64_t off = c->frag_offset;
1213 133 : if (off < c->value_header_len) {
1214 55 : int n = c->value_header_len - (int)off;
1215 55 : if (n > frag_len) n = frag_len;
1216 55 : memcpy(frag, c->value_header + off, (size_t)n);
1217 55 : copied += n;
1218 55 : off += n;
1219 55 : }
1220 133 : if (copied < frag_len) {
1221 133 : int64_t data_off = off - c->value_header_len;
1222 133 : memcpy(frag + copied, c->value_data + data_off, (size_t)(frag_len - copied));
1223 133 : }
1224 :
1225 133 : cloudsync_payload_context *payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
1226 133 : if (!payload) { cloudsync_memory_free(frag); return SQLITE_NOMEM; }
1227 266 : int rc = cloudsync_payload_encode_fragment_step(payload, data,
1228 133 : (const char *)sqlite3_column_text(c->src, 0), sqlite3_column_bytes(c->src, 0),
1229 133 : sqlite3_column_blob(c->src, 1), sqlite3_column_bytes(c->src, 1),
1230 133 : (const char *)sqlite3_column_text(c->src, 2), sqlite3_column_bytes(c->src, 2),
1231 133 : frag, frag_len,
1232 133 : sqlite3_column_int64(c->src, 4), sqlite3_column_int64(c->src, 5),
1233 133 : sqlite3_column_blob(c->src, 6), sqlite3_column_bytes(c->src, 6),
1234 133 : sqlite3_column_int64(c->src, 7), sqlite3_column_int64(c->src, 8),
1235 133 : c->frag_checksum, c->frag_total, c->frag_part, c->frag_count);
1236 133 : cloudsync_memory_free(frag);
1237 133 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1238 133 : rc = cloudsync_payload_encode_final(payload, data);
1239 133 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1240 : // A fragment chunk spends the window budget like any other. The fragment paths
1241 : // return before the ordinary builder's accounting, so without this a history made
1242 : // of oversized values never spends the budget and the cap never fires.
1243 133 : c->window_bytes += (int64_t)cloudsync_payload_context_bused(payload);
1244 133 : c->payload = cloudsync_payload_blob(payload, &c->payload_size, &c->rows);
1245 133 : cloudsync_memory_free(payload);
1246 133 : c->dbv_min = sqlite3_column_int64(c->src, 5);
1247 133 : c->dbv_max = c->dbv_min;
1248 133 : c->chunk_index++;
1249 133 : c->frag_offset += frag_len;
1250 133 : c->frag_part++;
1251 133 : if (c->frag_part >= c->frag_count) {
1252 55 : c->frag_active = false;
1253 55 : rc = payload_chunks_step_source(c);
1254 55 : }
1255 133 : return rc;
1256 133 : }
1257 :
1258 376 : static int payload_chunks_build_next(cloudsync_payload_chunks_cursor *c) {
1259 376 : cloudsync_context *data = c->vtab->data;
1260 376 : int rc = SQLITE_OK;
1261 376 : if (c->payload) { cloudsync_memory_free(c->payload); c->payload = NULL; }
1262 376 : c->payload_size = c->rows = c->dbv_min = c->dbv_max = 0;
1263 376 : if (c->frag_active) return payload_chunks_emit_fragment(c);
1264 298 : if (!c->has_row) { c->eof = true; return SQLITE_OK; }
1265 :
1266 246 : int max_size = cloudsync_payload_max_chunk_size(data);
1267 246 : size_t payload_header_size = 0;
1268 246 : cloudsync_payload_context_size(&payload_header_size);
1269 246 : cloudsync_payload_context *payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
1270 246 : if (!payload) return SQLITE_NOMEM;
1271 3431 : while (c->has_row) {
1272 : sqlite3_value *rowv[9];
1273 33820 : for (int i = 0; i < 9; ++i) rowv[i] = sqlite3_column_value(c->src, i);
1274 3382 : size_t row_size = pk_encode_size((dbvalue_t **)rowv, 9, 0, -1);
1275 3382 : if (row_size == SIZE_MAX) { cloudsync_memory_free(payload); return SQLITE_NOMEM; }
1276 :
1277 3382 : if ((int64_t)row_size + (int64_t)payload_header_size + CLOUDSYNC_PAYLOAD_CHUNK_SAFETY_MARGIN > max_size) {
1278 61 : if (cloudsync_payload_context_nrows(payload) > 0) break;
1279 55 : cloudsync_memory_free(payload);
1280 55 : rc = payload_chunks_begin_fragment(c, 0);
1281 55 : if (rc != SQLITE_OK) return rc;
1282 55 : return payload_chunks_emit_fragment(c);
1283 : }
1284 :
1285 3321 : if (cloudsync_payload_context_nrows(payload) > 0 && cloudsync_payload_context_bused(payload) + row_size > (size_t)max_size) break;
1286 : // Once the budget is spent, end the chunk at the first db_version boundary so
1287 : // the window can be capped there. Waiting for a chunk to happen to end on a
1288 : // boundary is not enough: when a transaction's rows and a chunk's capacity stay
1289 : // out of step, every chunk ends mid-version and the cap never fires at all.
1290 3377 : if (c->max_window_bytes > 0 && cloudsync_payload_context_nrows(payload) > 0 &&
1291 671 : sqlite3_column_int64(c->src, 5) != c->dbv_max &&
1292 141 : c->window_bytes + (int64_t)cloudsync_payload_context_bused(payload) >= c->max_window_bytes) break;
1293 3185 : rc = cloudsync_payload_encode_step(payload, data, 9, (dbvalue_t **)rowv);
1294 3185 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1295 3185 : int64_t dbv = sqlite3_column_int64(c->src, 5);
1296 3185 : if (cloudsync_payload_context_nrows(payload) == 1) c->dbv_min = dbv;
1297 3185 : c->dbv_max = dbv;
1298 3185 : rc = payload_chunks_step_source(c);
1299 3185 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1300 : }
1301 :
1302 191 : if (cloudsync_payload_context_nrows(payload) == 0) { cloudsync_memory_free(payload); c->eof = true; return SQLITE_OK; }
1303 191 : rc = cloudsync_payload_encode_final(payload, data);
1304 191 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1305 : // Measure the window in the unit max_size is expressed in -- encoded bytes before
1306 : // compression -- so a budget and a chunk size mean the same thing to a caller.
1307 191 : c->window_bytes += (int64_t)cloudsync_payload_context_bused(payload);
1308 191 : c->payload = cloudsync_payload_blob(payload, &c->payload_size, &c->rows);
1309 191 : cloudsync_memory_free(payload);
1310 191 : c->chunk_index++;
1311 191 : return SQLITE_OK;
1312 376 : }
1313 :
1314 : // Record the resume point a stateless caller passes back to continue after the
1315 : // chunk just built. Reads the source statement, which is positioned at the next
1316 : // unconsumed row (or the same row when a value is still mid-fragment). Must be
1317 : // called only after build_next produced a chunk (i.e. !eof).
1318 324 : static void payload_chunks_set_next_cursor(cloudsync_payload_chunks_cursor *c) {
1319 324 : if (c->frag_active) {
1320 : // Mid-value: resume the same row at the next byte offset.
1321 78 : c->next_dbv = sqlite3_column_int64(c->src, 5);
1322 78 : c->next_seq = sqlite3_column_int64(c->src, 8);
1323 78 : c->next_frag_offset = c->frag_offset;
1324 78 : c->is_final = false;
1325 324 : } else if (c->has_row) {
1326 : // Row boundary: the next chunk starts at the current (unconsumed) row.
1327 188 : c->next_dbv = sqlite3_column_int64(c->src, 5);
1328 188 : c->next_seq = sqlite3_column_int64(c->src, 8);
1329 188 : c->next_frag_offset = 0;
1330 188 : c->is_final = false;
1331 188 : } else {
1332 : // Stream exhausted: this was the last chunk of the window.
1333 58 : c->next_dbv = c->watermark;
1334 58 : c->next_seq = 0;
1335 58 : c->next_frag_offset = 0;
1336 58 : c->is_final = true;
1337 : }
1338 324 : }
1339 :
1340 : // Stop the scan once max_window_bytes is spent, at the first db_version boundary, and
1341 : // report the window as an ordinary complete stream ending at that reduced watermark.
1342 : // The caller then checkpoints there and asks again, so preparation is bounded without
1343 : // any new resumable state.
1344 : //
1345 : // Two conditions are load-bearing. Never cap mid-value (frag_active) or mid-db_version
1346 : // (next_dbv == dbv_max): the receive cursor must land on a complete db_version or the
1347 : // next request skips the unapplied remainder, since it resumes with db_version > since
1348 : // and no seq. And because the boundary test is what stops us, a db_version larger than
1349 : // the whole budget is still emitted in full -- otherwise a window could end up empty
1350 : // and the drain would never advance at all.
1351 324 : static void payload_chunks_apply_window_cap(cloudsync_payload_chunks_cursor *c) {
1352 324 : if (c->max_window_bytes <= 0 || c->eof) return;
1353 138 : if (c->is_final || c->frag_active) return;
1354 114 : if (c->window_bytes < c->max_window_bytes) return;
1355 110 : if (c->next_dbv == c->dbv_max) return; // still inside a db_version
1356 :
1357 106 : c->window_capped = true;
1358 106 : c->is_final = true;
1359 106 : c->watermark = c->dbv_max; // the window the caller checkpoints at
1360 106 : c->next_dbv = c->watermark;
1361 106 : c->next_seq = 0;
1362 106 : c->next_frag_offset = 0;
1363 324 : }
1364 :
1365 376 : static int payload_chunks_advance(cloudsync_payload_chunks_cursor *c) {
1366 376 : int rc = payload_chunks_build_next(c);
1367 376 : if (rc == SQLITE_OK && !c->eof) {
1368 324 : payload_chunks_set_next_cursor(c);
1369 324 : payload_chunks_apply_window_cap(c);
1370 324 : }
1371 376 : return rc;
1372 : }
1373 :
1374 187 : static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const char *idxstr, int argc, sqlite3_value **argv) {
1375 187 : UNUSED_PARAMETER(idxstr); UNUSED_PARAMETER(argc);
1376 187 : cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
1377 187 : cloudsync_context *data = c->vtab->data;
1378 187 : if (!cloudsync_context_is_initialized(data)) {
1379 1 : c->vtab->base.zErrMsg = sqlite3_mprintf(
1380 : "cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') "
1381 : "to enable sync on a table before querying cloudsync_payload_chunks.");
1382 1 : return SQLITE_ERROR;
1383 : }
1384 186 : if (c->src) { sqlite3_finalize(c->src); c->src = NULL; }
1385 186 : if (c->payload) { cloudsync_memory_free(c->payload); c->payload = NULL; }
1386 : // Contract: all per-scan state that can be bulk-reset here must live at or
1387 : // after eof. Fields before eof are cursor lifetime state preserved across
1388 : // xFilter calls.
1389 186 : memset(&c->eof, 0, sizeof(*c) - offsetof(cloudsync_payload_chunks_cursor, eof));
1390 :
1391 186 : int argi = 0;
1392 186 : int64_t since = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_SEND_DBVERSION);
1393 186 : const void *site_id = NULL;
1394 186 : int site_id_len = 0;
1395 186 : bool site_id_given = false;
1396 186 : int64_t until = 0;
1397 186 : bool exclude = false;
1398 : // Positional resume cursor (cols 15..17): when resume_db_version is bound the
1399 : // scan starts at (resume_db_version, resume_seq) inclusive and the first chunk
1400 : // resumes a mid-value fragment at resume_frag_offset, instead of replaying the
1401 : // whole window from `since`. Lets a stateless /check page the stream with an
1402 : // O(1) seek per call and no spool table.
1403 186 : bool positional = false;
1404 186 : int64_t resume_dbv = 0, resume_seq = 0, resume_frag = 0;
1405 186 : if (idxnum & 1) since = sqlite3_value_int64(argv[argi++]);
1406 186 : if (idxnum & 2) {
1407 119 : if (sqlite3_value_type(argv[argi]) != SQLITE_NULL) {
1408 2 : site_id = sqlite3_value_blob(argv[argi]);
1409 2 : site_id_len = sqlite3_value_bytes(argv[argi]);
1410 2 : site_id_given = true;
1411 2 : }
1412 119 : argi++;
1413 119 : }
1414 186 : if (idxnum & 4) until = sqlite3_value_int64(argv[argi++]);
1415 186 : if (idxnum & 8) exclude = (sqlite3_value_int(argv[argi++]) != 0);
1416 : // An explicit NULL means "not given", as it already does for site_id. Callers must
1417 : // pass NULLs to reach a later argument, and treating one as a resume point of 0
1418 : // would silently restart the scan at the beginning of the window.
1419 186 : if (idxnum & 16) {
1420 124 : if (sqlite3_value_type(argv[argi]) != SQLITE_NULL) {
1421 14 : resume_dbv = sqlite3_value_int64(argv[argi]);
1422 14 : positional = true;
1423 14 : }
1424 124 : argi++;
1425 124 : }
1426 186 : if (idxnum & 32) resume_seq = sqlite3_value_int64(argv[argi++]);
1427 186 : if (idxnum & 64) resume_frag = sqlite3_value_int64(argv[argi++]);
1428 186 : if (idxnum & 128) {
1429 114 : int64_t cap = sqlite3_value_int64(argv[argi++]);
1430 114 : c->max_window_bytes = (cap > 0) ? cap : 0; // <= 0 means no cap
1431 114 : }
1432 : // Budget already spent by earlier calls of this window. The per-scan reset above
1433 : // cleared window_bytes, so a caller paging one chunk at a time seeds it here; a
1434 : // caller draining the window in one scan leaves it at 0 and it accumulates.
1435 186 : if (idxnum & 256) {
1436 6 : int64_t spent = sqlite3_value_int64(argv[argi++]);
1437 6 : c->window_bytes = (spent > 0) ? spent : 0;
1438 6 : }
1439 :
1440 : // Resolve the site filter:
1441 : // exclude=true -> all sites except filter_site_id (CHECK path); site required
1442 : // filter given -> only that site
1443 : // default -> local site (send path, unchanged)
1444 : const char *site_op;
1445 186 : if (exclude) {
1446 2 : if (!site_id_given) {
1447 1 : c->vtab->base.zErrMsg = sqlite3_mprintf(
1448 : "cloudsync_payload_chunks: exclude_filter_site_id requires a non-NULL site_id");
1449 1 : return SQLITE_ERROR;
1450 : }
1451 1 : site_op = "<>";
1452 1 : } else {
1453 184 : site_op = "=";
1454 184 : if (!site_id_given) { site_id = cloudsync_siteid(data); site_id_len = UUID_LEN; }
1455 : }
1456 :
1457 185 : if (until == 0) {
1458 161 : char *mxsql = sqlite3_mprintf(
1459 161 : "SELECT COALESCE(MAX(db_version),0) FROM cloudsync_changes WHERE site_id%s?", site_op);
1460 161 : if (!mxsql) return SQLITE_NOMEM;
1461 161 : sqlite3_stmt *mx = NULL;
1462 161 : int rc = sqlite3_prepare_v2(c->vtab->db, mxsql, -1, &mx, NULL);
1463 161 : sqlite3_free(mxsql);
1464 161 : if (rc != SQLITE_OK) return rc;
1465 161 : sqlite3_bind_blob(mx, 1, site_id, site_id_len, SQLITE_TRANSIENT);
1466 : // MAX() yields exactly one row, so anything else is an error: propagate
1467 : // it instead of silently scanning an empty window with until=0
1468 161 : rc = sqlite3_step(mx);
1469 161 : if (rc == SQLITE_ROW) until = sqlite3_column_int64(mx, 0);
1470 161 : sqlite3_finalize(mx);
1471 161 : if (rc != SQLITE_ROW) {
1472 0 : if (c->vtab->base.zErrMsg) sqlite3_free(c->vtab->base.zErrMsg);
1473 0 : c->vtab->base.zErrMsg = sqlite3_mprintf("%s", sqlite3_errmsg(c->vtab->db));
1474 0 : return rc == SQLITE_DONE ? SQLITE_ERROR : rc;
1475 : }
1476 161 : }
1477 185 : c->watermark = until;
1478 :
1479 : // Window upper bound is always `until`. The lower bound is either the legacy
1480 : // exclusive `since` (db_version > since) or the inclusive positional cursor
1481 : // (db_version, seq) >= (resume_dbv, resume_seq).
1482 : char *sql;
1483 185 : if (positional) {
1484 : // The redundant db_version>=? is what makes the resume a seek, and it is
1485 : // load-bearing: SQLite derives a range from a disjunction only when both arms
1486 : // compare against the same value, and these two arms carry distinct
1487 : // parameters. Without it cloudsync_changes' xBestIndex is offered no lower
1488 : // bound at all and every call replays the window from the start.
1489 14 : sql = sqlite3_mprintf(
1490 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1491 : "FROM cloudsync_changes WHERE db_version<=? AND site_id%s? AND db_version>=? AND "
1492 : "(db_version>? OR (db_version=? AND seq>=?)) ORDER BY db_version, seq ASC",
1493 14 : site_op);
1494 14 : } else {
1495 171 : sql = sqlite3_mprintf(
1496 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1497 : "FROM cloudsync_changes WHERE db_version>? AND site_id%s? AND db_version<=? ORDER BY db_version, seq ASC",
1498 171 : site_op);
1499 : }
1500 185 : if (!sql) return SQLITE_NOMEM;
1501 185 : int rc = sqlite3_prepare_v2(c->vtab->db, sql, -1, &c->src, NULL);
1502 185 : sqlite3_free(sql);
1503 185 : if (rc != SQLITE_OK) return rc;
1504 185 : if (positional) {
1505 14 : sqlite3_bind_int64(c->src, 1, until);
1506 14 : sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT);
1507 14 : sqlite3_bind_int64(c->src, 3, resume_dbv);
1508 14 : sqlite3_bind_int64(c->src, 4, resume_dbv);
1509 14 : sqlite3_bind_int64(c->src, 5, resume_dbv);
1510 14 : sqlite3_bind_int64(c->src, 6, resume_seq);
1511 14 : } else {
1512 171 : sqlite3_bind_int64(c->src, 1, since);
1513 171 : sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT);
1514 171 : sqlite3_bind_int64(c->src, 3, until);
1515 : }
1516 185 : rc = payload_chunks_step_source(c);
1517 185 : if (rc != SQLITE_OK) return rc;
1518 : // Resuming inside a value that was fragmented across chunks: the first row is
1519 : // that value; re-establish the fragment plan and skip to resume_frag.
1520 185 : if (positional && resume_frag > 0 && c->has_row) {
1521 6 : rc = payload_chunks_begin_fragment(c, resume_frag);
1522 6 : if (rc != SQLITE_OK) return rc;
1523 6 : }
1524 185 : return payload_chunks_advance(c);
1525 187 : }
1526 :
1527 296 : static int payload_chunks_next(sqlite3_vtab_cursor *cursor) {
1528 296 : cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
1529 : // A capped window ends the scan: the rows past it belong to the next window, and
1530 : // emitting them here would contradict the reduced watermark already reported.
1531 296 : if (c->window_capped) { c->eof = true; return SQLITE_OK; }
1532 191 : return payload_chunks_advance(c);
1533 296 : }
1534 :
1535 481 : static int payload_chunks_eof(sqlite3_vtab_cursor *cursor) {
1536 481 : return ((cloudsync_payload_chunks_cursor *)cursor)->eof;
1537 : }
1538 :
1539 1378 : static int payload_chunks_column(sqlite3_vtab_cursor *cursor, sqlite3_context *ctx, int col) {
1540 1378 : cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
1541 1378 : switch (col) {
1542 129 : case 0: sqlite3_result_blob64(ctx, c->payload, (sqlite3_uint64)c->payload_size, SQLITE_TRANSIENT); break;
1543 63 : case 1: sqlite3_result_int(ctx, c->chunk_index - 1); break;
1544 173 : case 2: sqlite3_result_int64(ctx, c->payload_size); break;
1545 226 : case 3: sqlite3_result_int64(ctx, c->rows); break;
1546 141 : case 4: sqlite3_result_int64(ctx, c->dbv_min); break;
1547 9 : case 5: sqlite3_result_int64(ctx, c->dbv_max); break;
1548 162 : case 6: sqlite3_result_int64(ctx, c->watermark); break;
1549 16 : case 11: sqlite3_result_int64(ctx, c->next_dbv); break;
1550 16 : case 12: sqlite3_result_int64(ctx, c->next_seq); break;
1551 16 : case 13: sqlite3_result_int64(ctx, c->next_frag_offset); break;
1552 257 : case 14: sqlite3_result_int(ctx, c->is_final ? 1 : 0); break;
1553 164 : case 15: sqlite3_result_int(ctx, c->window_capped ? 1 : 0); break;
1554 6 : case 16: sqlite3_result_int64(ctx, c->window_bytes); break;
1555 0 : default: sqlite3_result_null(ctx); break;
1556 : }
1557 1378 : return SQLITE_OK;
1558 : }
1559 :
1560 0 : static int payload_chunks_rowid(sqlite3_vtab_cursor *cursor, sqlite3_int64 *rowid) {
1561 0 : *rowid = ((cloudsync_payload_chunks_cursor *)cursor)->chunk_index;
1562 0 : return SQLITE_OK;
1563 : }
1564 :
1565 : static sqlite3_module cloudsync_payload_chunks_module = {
1566 : /* iVersion */ 0,
1567 : /* xCreate */ NULL,
1568 : /* xConnect */ payload_chunks_connect,
1569 : /* xBestIndex */ payload_chunks_best_index,
1570 : /* xDisconnect */ payload_chunks_disconnect,
1571 : /* xDestroy */ NULL,
1572 : /* xOpen */ payload_chunks_open,
1573 : /* xClose */ payload_chunks_close,
1574 : /* xFilter */ payload_chunks_filter,
1575 : /* xNext */ payload_chunks_next,
1576 : /* xEof */ payload_chunks_eof,
1577 : /* xColumn */ payload_chunks_column,
1578 : /* xRowid */ payload_chunks_rowid,
1579 : /* xUpdate */ NULL,
1580 : /* xBegin */ NULL,
1581 : /* xSync */ NULL,
1582 : /* xCommit */ NULL,
1583 : /* xRollback */ NULL,
1584 : /* xFindMethod */ NULL,
1585 : /* xRename */ NULL,
1586 : /* xSavepoint */ NULL,
1587 : /* xRelease */ NULL,
1588 : /* xRollbackTo */ NULL,
1589 : /* xShadowName */ NULL,
1590 : /* xIntegrity */ NULL
1591 : };
1592 :
1593 1060 : static int payload_estimated_size_add(sqlite3_int64 *acc, sqlite3_int64 value) {
1594 1060 : if (value < 0 || *acc > INT64_MAX - value) return SQLITE_TOOBIG;
1595 1060 : *acc += value;
1596 1060 : return SQLITE_OK;
1597 1060 : }
1598 :
1599 6 : static int payload_blob_checked_estimate(sqlite3 *db, const void *site_id, int site_id_len,
1600 : sqlite3_int64 since, sqlite3_int64 since_seq,
1601 : bool exclude, sqlite3_int64 *estimated,
1602 : sqlite3_int64 *watermark) {
1603 6 : sqlite3_stmt *stmt = NULL;
1604 6 : sqlite3_stmt *mx = NULL;
1605 6 : int rc = SQLITE_OK;
1606 6 : sqlite3_int64 until = 0;
1607 6 : size_t header_size = 0;
1608 6 : bool has_rows = false;
1609 6 : const char *site_op = exclude ? "<>" : "=";
1610 6 : *estimated = 0;
1611 :
1612 6 : char *mxsql = sqlite3_mprintf(
1613 6 : "SELECT COALESCE(MAX(db_version),0) FROM cloudsync_changes WHERE site_id%s?", site_op);
1614 6 : if (!mxsql) return SQLITE_NOMEM;
1615 6 : rc = sqlite3_prepare_v2(db, mxsql, -1, &mx, NULL);
1616 6 : sqlite3_free(mxsql);
1617 6 : if (rc != SQLITE_OK) goto error;
1618 6 : sqlite3_bind_blob(mx, 1, site_id, site_id_len, SQLITE_TRANSIENT);
1619 6 : rc = sqlite3_step(mx);
1620 6 : if (rc == SQLITE_ROW) until = sqlite3_column_int64(mx, 0);
1621 0 : else if (rc != SQLITE_DONE) goto error;
1622 6 : sqlite3_finalize(mx);
1623 6 : mx = NULL;
1624 6 : if (watermark) *watermark = until;
1625 :
1626 6 : if (until < since) return SQLITE_OK;
1627 :
1628 4 : char *sql = sqlite3_mprintf(
1629 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1630 : "FROM cloudsync_changes WHERE (db_version>? OR (db_version=? AND seq>?)) "
1631 : "AND site_id%s? AND db_version<=? ORDER BY db_version, seq ASC",
1632 4 : site_op);
1633 4 : if (!sql) return SQLITE_NOMEM;
1634 4 : rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
1635 4 : sqlite3_free(sql);
1636 4 : if (rc != SQLITE_OK) goto error;
1637 4 : sqlite3_bind_int64(stmt, 1, since);
1638 4 : sqlite3_bind_int64(stmt, 2, since);
1639 4 : sqlite3_bind_int64(stmt, 3, since_seq);
1640 4 : sqlite3_bind_blob(stmt, 4, site_id, site_id_len, SQLITE_TRANSIENT);
1641 4 : sqlite3_bind_int64(stmt, 5, until);
1642 :
1643 4 : cloudsync_payload_context_size(&header_size);
1644 1060 : while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
1645 : sqlite3_value *rowv[9];
1646 10560 : for (int i = 0; i < 9; ++i) rowv[i] = sqlite3_column_value(stmt, i);
1647 1056 : size_t row_size = pk_encode_size((dbvalue_t **)rowv, 9, 0, -1);
1648 1056 : if (row_size == SIZE_MAX || row_size > (size_t)INT64_MAX) {
1649 0 : rc = SQLITE_TOOBIG;
1650 0 : goto error;
1651 : }
1652 1056 : if (!has_rows) {
1653 4 : if (header_size > (size_t)INT64_MAX) {
1654 0 : rc = SQLITE_TOOBIG;
1655 0 : goto error;
1656 : }
1657 4 : rc = payload_estimated_size_add(estimated, (sqlite3_int64)header_size);
1658 4 : if (rc != SQLITE_OK) goto error;
1659 4 : has_rows = true;
1660 4 : }
1661 1056 : rc = payload_estimated_size_add(estimated, (sqlite3_int64)row_size);
1662 1056 : if (rc != SQLITE_OK) goto error;
1663 : }
1664 4 : if (rc != SQLITE_DONE) goto error;
1665 4 : sqlite3_finalize(stmt);
1666 4 : return SQLITE_OK;
1667 :
1668 : error:
1669 0 : if (stmt) sqlite3_finalize(stmt);
1670 0 : if (mx) sqlite3_finalize(mx);
1671 0 : return rc;
1672 6 : }
1673 :
1674 7 : void dbsync_payload_blob_checked(sqlite3_context *context, int argc, sqlite3_value **argv) {
1675 : DEBUG_FUNCTION("cloudsync_payload_blob_checked");
1676 7 : UNUSED_PARAMETER(argc);
1677 :
1678 7 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1679 7 : sqlite3 *db = sqlite3_context_db_handle(context);
1680 7 : sqlite3_stmt *stmt = NULL;
1681 7 : cloudsync_payload_context *payload = NULL;
1682 7 : int rc = SQLITE_OK;
1683 7 : sqlite3_int64 since = 0;
1684 7 : sqlite3_int64 since_seq = 0;
1685 7 : sqlite3_int64 max_estimated_size = 0;
1686 7 : sqlite3_int64 estimated = 0;
1687 7 : sqlite3_int64 watermark = 0;
1688 7 : const void *site_id = NULL;
1689 7 : int site_id_len = 0;
1690 7 : bool exclude = false;
1691 :
1692 7 : if (sqlite3_value_type(argv[0]) != SQLITE_NULL) since = sqlite3_value_int64(argv[0]);
1693 7 : if (sqlite3_value_type(argv[1]) != SQLITE_NULL) since_seq = sqlite3_value_int64(argv[1]);
1694 7 : if (sqlite3_value_type(argv[2]) != SQLITE_NULL) {
1695 3 : site_id = sqlite3_value_blob(argv[2]);
1696 3 : site_id_len = sqlite3_value_bytes(argv[2]);
1697 3 : }
1698 7 : exclude = sqlite3_value_type(argv[3]) != SQLITE_NULL && sqlite3_value_int(argv[3]) != 0;
1699 7 : if (sqlite3_value_type(argv[4]) == SQLITE_NULL || sqlite3_value_int64(argv[4]) <= 0) {
1700 0 : sqlite3_result_error(context, "cloudsync_payload_blob_checked: max_estimated_payload_size must be positive", -1);
1701 0 : return;
1702 : }
1703 7 : max_estimated_size = sqlite3_value_int64(argv[4]);
1704 :
1705 7 : if (exclude && !site_id) {
1706 1 : sqlite3_result_error(context,
1707 : "cloudsync_payload_blob_checked: exclude_filter_site_id requires a non-NULL site_id", -1);
1708 1 : return;
1709 : }
1710 6 : if (!exclude && !site_id) {
1711 3 : site_id = cloudsync_siteid(data);
1712 3 : site_id_len = UUID_LEN;
1713 3 : }
1714 :
1715 6 : rc = payload_blob_checked_estimate(db, site_id, site_id_len, since, since_seq, exclude, &estimated, &watermark);
1716 6 : if (rc != SQLITE_OK) goto error;
1717 6 : if (estimated == 0) {
1718 2 : sqlite3_result_null(context);
1719 2 : return;
1720 : }
1721 4 : if (estimated > max_estimated_size) {
1722 2 : dbsync_set_error(context,
1723 : CLOUDSYNC_ERRCODE_PAYLOAD_TOO_LARGE "cloudsync_payload_blob_checked: estimated payload size %" PRId64 " exceeds max_estimated_payload_size %" PRId64,
1724 1 : (int64_t)estimated, (int64_t)max_estimated_size);
1725 1 : return;
1726 : }
1727 :
1728 3 : const char *site_op = exclude ? "<>" : "=";
1729 3 : char *sql = sqlite3_mprintf(
1730 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1731 : "FROM cloudsync_changes WHERE (db_version>? OR (db_version=? AND seq>?)) "
1732 : "AND site_id%s? AND db_version<=? ORDER BY db_version, seq ASC",
1733 3 : site_op);
1734 3 : if (!sql) { rc = SQLITE_NOMEM; goto error; }
1735 3 : rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
1736 3 : sqlite3_free(sql);
1737 3 : if (rc != SQLITE_OK) goto error;
1738 3 : sqlite3_bind_int64(stmt, 1, since);
1739 3 : sqlite3_bind_int64(stmt, 2, since);
1740 3 : sqlite3_bind_int64(stmt, 3, since_seq);
1741 3 : sqlite3_bind_blob(stmt, 4, site_id, site_id_len, SQLITE_TRANSIENT);
1742 3 : sqlite3_bind_int64(stmt, 5, watermark);
1743 :
1744 3 : payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
1745 3 : if (!payload) { rc = SQLITE_NOMEM; goto error; }
1746 :
1747 533 : while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
1748 : sqlite3_value *rowv[9];
1749 5300 : for (int i = 0; i < 9; ++i) rowv[i] = sqlite3_column_value(stmt, i);
1750 530 : rc = cloudsync_payload_encode_step(payload, data, 9, (dbvalue_t **)rowv);
1751 530 : if (rc != SQLITE_OK) goto error;
1752 : }
1753 3 : if (rc != SQLITE_DONE) goto error;
1754 3 : sqlite3_finalize(stmt);
1755 3 : stmt = NULL;
1756 :
1757 3 : rc = cloudsync_payload_encode_final(payload, data);
1758 3 : if (rc != SQLITE_OK) goto error;
1759 3 : int64_t blob_size = 0;
1760 3 : char *blob = cloudsync_payload_blob(payload, &blob_size, NULL);
1761 3 : if (!blob) {
1762 0 : sqlite3_result_null(context);
1763 0 : cloudsync_payload_context_free(payload);
1764 0 : } else {
1765 3 : sqlite3_result_blob64(context, blob, (sqlite3_uint64)blob_size, cloudsync_memory_free);
1766 3 : cloudsync_memory_free(payload);
1767 : }
1768 3 : return;
1769 :
1770 : error:
1771 0 : if (stmt) sqlite3_finalize(stmt);
1772 0 : if (payload) cloudsync_payload_context_free(payload);
1773 0 : if (rc == SQLITE_NOMEM) sqlite3_result_error_nomem(context);
1774 0 : else if (rc == SQLITE_TOOBIG) sqlite3_result_error(context, CLOUDSYNC_ERRCODE_PAYLOAD_TOO_LARGE "cloudsync_payload_blob_checked: payload estimate is too large", -1);
1775 0 : else sqlite3_result_error(context, sqlite3_errmsg(db), -1);
1776 7 : }
1777 :
1778 : #ifdef CLOUDSYNC_DESKTOP_OS
1779 0 : void dbsync_payload_save (sqlite3_context *context, int argc, sqlite3_value **argv) {
1780 : DEBUG_FUNCTION("dbsync_payload_save");
1781 :
1782 : // sanity check argument
1783 0 : if (database_value_type(argv[0]) != SQLITE_TEXT) {
1784 0 : sqlite3_result_error(context, "Unable to retrieve file path.", -1);
1785 0 : return;
1786 : }
1787 :
1788 : // retrieve full path to file
1789 0 : const char *payload_path = (const char *)database_value_text(argv[0]);
1790 :
1791 : // retrieve global context
1792 0 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1793 :
1794 0 : int blob_size = 0;
1795 0 : int rc = cloudsync_payload_save(data, payload_path, &blob_size);
1796 0 : if (rc == SQLITE_OK) {
1797 : // if OK then returns blob size
1798 0 : sqlite3_result_int64(context, (sqlite3_int64)blob_size);
1799 0 : return;
1800 : }
1801 :
1802 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
1803 0 : sqlite3_result_error_code(context, rc);
1804 0 : }
1805 :
1806 0 : void dbsync_payload_load (sqlite3_context *context, int argc, sqlite3_value **argv) {
1807 : DEBUG_FUNCTION("dbsync_payload_load");
1808 :
1809 : // sanity check argument
1810 0 : if (database_value_type(argv[0]) != SQLITE_TEXT) {
1811 0 : sqlite3_result_error(context, "Unable to retrieve file path.", -1);
1812 0 : return;
1813 : }
1814 :
1815 : // retrieve full path to file
1816 0 : const char *path = (const char *)database_value_text(argv[0]);
1817 :
1818 0 : int64_t payload_size = 0;
1819 0 : char *payload = cloudsync_file_read(path, &payload_size);
1820 0 : if (!payload) {
1821 0 : if (payload_size < 0) {
1822 0 : sqlite3_result_error(context, "Unable to read payload from file path.", -1);
1823 0 : sqlite3_result_error_code(context, SQLITE_IOERR);
1824 0 : return;
1825 : }
1826 : // no rows affected but no error either
1827 0 : sqlite3_result_int(context, 0);
1828 0 : return;
1829 : }
1830 :
1831 0 : int nrows = 0;
1832 0 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1833 : // File-based load applies a complete monolithic payload: legacy last-applied
1834 : // checkpoint (ends on a db_version boundary, so it is safe).
1835 0 : int rc = cloudsync_payload_apply (data, payload, (int)payload_size, &nrows, CLOUDSYNC_CHECKPOINT_LAST_APPLIED, 0);
1836 0 : if (payload) cloudsync_memory_free(payload);
1837 :
1838 0 : if (rc != SQLITE_OK) {
1839 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
1840 0 : sqlite3_result_error_code(context, rc);
1841 0 : return;
1842 : }
1843 :
1844 : // returns number of applied rows
1845 0 : sqlite3_result_int(context, nrows);
1846 0 : }
1847 : #endif
1848 :
1849 : // MARK: - Register -
1850 :
1851 32340 : int dbsync_register_with_flags (sqlite3 *db, const char *name, void (*xfunc)(sqlite3_context*,int,sqlite3_value**), void (*xstep)(sqlite3_context*,int,sqlite3_value**), void (*xfinal)(sqlite3_context*), int nargs, int flags, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
1852 :
1853 32340 : int rc = sqlite3_create_function_v2(db, name, nargs, flags, ctx, xfunc, xstep, xfinal, ctx_free);
1854 :
1855 32340 : if (rc != SQLITE_OK) {
1856 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("Error creating function %s: %s", name, sqlite3_errmsg(db));
1857 0 : return rc;
1858 : }
1859 32340 : return SQLITE_OK;
1860 32340 : }
1861 :
1862 26950 : int dbsync_register (sqlite3 *db, const char *name, void (*xfunc)(sqlite3_context*,int,sqlite3_value**), void (*xstep)(sqlite3_context*,int,sqlite3_value**), void (*xfinal)(sqlite3_context*), int nargs, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
1863 26950 : const int FLAGS_VOLATILE = SQLITE_UTF8;
1864 : DEBUG_DBFUNCTION("dbsync_register %s", name);
1865 26950 : return dbsync_register_with_flags(db, name, xfunc, xstep, xfinal, nargs, FLAGS_VOLATILE, pzErrMsg, ctx, ctx_free);
1866 : }
1867 :
1868 26180 : int dbsync_register_function (sqlite3 *db, const char *name, void (*xfunc)(sqlite3_context*,int,sqlite3_value**), int nargs, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
1869 : DEBUG_DBFUNCTION("dbsync_register_function %s", name);
1870 26180 : return dbsync_register(db, name, xfunc, NULL, NULL, nargs, pzErrMsg, ctx, ctx_free);
1871 : }
1872 :
1873 2310 : int dbsync_register_pure_function (sqlite3 *db, const char *name, void (*xfunc)(sqlite3_context*,int,sqlite3_value**), int nargs, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
1874 2310 : const int FLAGS_PURE = SQLITE_UTF8 | SQLITE_INNOCUOUS | SQLITE_DETERMINISTIC;
1875 : DEBUG_DBFUNCTION("dbsync_register_pure_function %s", name);
1876 2310 : return dbsync_register_with_flags(db, name, xfunc, NULL, NULL, nargs, FLAGS_PURE, pzErrMsg, ctx, ctx_free);
1877 : }
1878 :
1879 2310 : int dbsync_register_trigger_function (sqlite3 *db, const char *name, void (*xfunc)(sqlite3_context*,int,sqlite3_value**), int nargs, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
1880 2310 : const int FLAGS_TRIGGER = SQLITE_UTF8 | SQLITE_INNOCUOUS;
1881 : DEBUG_DBFUNCTION("dbsync_register_trigger_function %s", name);
1882 2310 : return dbsync_register_with_flags(db, name, xfunc, NULL, NULL, nargs, FLAGS_TRIGGER, pzErrMsg, ctx, ctx_free);
1883 : }
1884 :
1885 770 : int dbsync_register_aggregate (sqlite3 *db, const char *name, void (*xstep)(sqlite3_context*,int,sqlite3_value**), void (*xfinal)(sqlite3_context*), int nargs, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
1886 : DEBUG_DBFUNCTION("dbsync_register_aggregate %s", name);
1887 770 : return dbsync_register(db, name, NULL, xstep, xfinal, nargs, pzErrMsg, ctx, ctx_free);
1888 : }
1889 :
1890 770 : int dbsync_register_trigger_aggregate (sqlite3 *db, const char *name, void (*xstep)(sqlite3_context*,int,sqlite3_value**), void (*xfinal)(sqlite3_context*), int nargs, char **pzErrMsg, void *ctx, void (*ctx_free)(void *)) {
1891 770 : const int FLAGS_TRIGGER = SQLITE_UTF8 | SQLITE_INNOCUOUS;
1892 : DEBUG_DBFUNCTION("dbsync_register_trigger_aggregate %s", name);
1893 770 : return dbsync_register_with_flags(db, name, NULL, xstep, xfinal, nargs, FLAGS_TRIGGER, pzErrMsg, ctx, ctx_free);
1894 : }
1895 :
1896 : // MARK: - Block-level LWW -
1897 :
1898 603 : void dbsync_text_materialize (sqlite3_context *context, int argc, sqlite3_value **argv) {
1899 : DEBUG_FUNCTION("cloudsync_text_materialize");
1900 :
1901 : // argv[0] -> table name
1902 : // argv[1] -> column name
1903 : // argv[2..N] -> primary key values
1904 :
1905 603 : if (argc < 3) {
1906 0 : sqlite3_result_error(context, "cloudsync_text_materialize requires at least 3 arguments: table, column, pk...", -1);
1907 0 : return;
1908 : }
1909 :
1910 603 : const char *table_name = (const char *)database_value_text(argv[0]);
1911 603 : const char *col_name = (const char *)database_value_text(argv[1]);
1912 603 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1913 :
1914 603 : cloudsync_table_context *table = table_lookup(data, table_name);
1915 603 : if (!table) {
1916 0 : dbsync_set_error(context, "Unable to retrieve table name %s in cloudsync_text_materialize.", table_name);
1917 0 : return;
1918 : }
1919 :
1920 603 : int col_idx = table_col_index(table, col_name);
1921 603 : if (col_idx < 0 || table_col_algo(table, col_idx) != col_algo_block) {
1922 0 : dbsync_set_error(context, "Column %s in table %s is not configured as block-level.", col_name, table_name);
1923 0 : return;
1924 : }
1925 :
1926 : // Encode primary keys
1927 603 : int npks = table_count_pks(table);
1928 603 : if (argc - 2 != npks) {
1929 0 : sqlite3_result_error(context, "Wrong number of primary key values for cloudsync_text_materialize.", -1);
1930 0 : return;
1931 : }
1932 :
1933 : char buffer[1024];
1934 603 : size_t pklen = sizeof(buffer);
1935 603 : char *pk = pk_encode_prikey((dbvalue_t **)&argv[2], npks, buffer, &pklen);
1936 603 : if (!pk || pk == PRIKEY_NULL_CONSTRAINT_ERROR) {
1937 0 : sqlite3_result_error(context, "Failed to encode primary key(s).", -1);
1938 0 : return;
1939 : }
1940 :
1941 : // Materialize the column
1942 603 : int rc = block_materialize_column(data, table, pk, (int)pklen, col_name);
1943 603 : if (rc != DBRES_OK) {
1944 309 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
1945 309 : sqlite3_result_error_code(context, rc);
1946 309 : } else {
1947 294 : sqlite3_result_int(context, 1);
1948 : }
1949 :
1950 603 : if (pk != buffer) cloudsync_memory_free(pk);
1951 603 : }
1952 :
1953 : // MARK: - Row Filter -
1954 :
1955 21 : void dbsync_set_filter (sqlite3_context *context, int argc, sqlite3_value **argv) {
1956 : DEBUG_FUNCTION("cloudsync_set_filter");
1957 :
1958 21 : const char *tbl = (const char *)database_value_text(argv[0]);
1959 21 : const char *filter_expr = (const char *)database_value_text(argv[1]);
1960 21 : if (!tbl || !filter_expr) {
1961 0 : dbsync_set_error(context, "cloudsync_set_filter: table and filter expression required");
1962 0 : return;
1963 : }
1964 :
1965 21 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1966 :
1967 : // Guard against calling set_filter before the target table has been set
1968 : // up for sync: without this, we'd hit "no such table:
1969 : // cloudsync_table_settings" or "no such table: main.<tbl>" deep inside
1970 : // the trigger recreation path, which is not actionable.
1971 21 : if (!cloudsync_context_is_initialized(data) || !table_lookup(data, tbl)) {
1972 2 : dbsync_set_error(context,
1973 : "cloudsync_set_filter: table '%s' is not configured for sync. "
1974 1 : "Call SELECT cloudsync_init('%s') first.", tbl, tbl);
1975 1 : return;
1976 : }
1977 :
1978 : // Store filter in table settings
1979 20 : dbutils_table_settings_set_key_value(data, tbl, "*", "filter", filter_expr);
1980 :
1981 : // Read current algo
1982 20 : table_algo algo = dbutils_table_settings_get_algo(data, tbl);
1983 20 : if (algo == table_algo_none) algo = table_algo_crdt_cls;
1984 :
1985 : // Drop and recreate triggers with the filter
1986 20 : database_delete_triggers(data, tbl);
1987 20 : int rc = database_create_triggers(data, tbl, algo, filter_expr);
1988 20 : if (rc != DBRES_OK) {
1989 0 : dbsync_set_error(context, "cloudsync_set_filter: error recreating triggers");
1990 0 : sqlite3_result_error_code(context, rc);
1991 0 : return;
1992 : }
1993 :
1994 : // Clean and refill metatable with the new filter
1995 20 : rc = cloudsync_reset_metatable(data, tbl);
1996 20 : if (rc != DBRES_OK) {
1997 0 : dbsync_set_error(context, "cloudsync_set_filter: error resetting metatable");
1998 0 : sqlite3_result_error_code(context, rc);
1999 0 : return;
2000 : }
2001 :
2002 20 : sqlite3_result_int(context, 1);
2003 21 : }
2004 :
2005 5 : void dbsync_clear_filter (sqlite3_context *context, int argc, sqlite3_value **argv) {
2006 : DEBUG_FUNCTION("cloudsync_clear_filter");
2007 :
2008 5 : const char *tbl = (const char *)database_value_text(argv[0]);
2009 5 : if (!tbl) {
2010 0 : dbsync_set_error(context, "cloudsync_clear_filter: table name required");
2011 0 : return;
2012 : }
2013 :
2014 5 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
2015 :
2016 : // Guard against calling clear_filter before the target table has been set
2017 : // up for sync — see dbsync_set_filter for the same rationale.
2018 5 : if (!cloudsync_context_is_initialized(data) || !table_lookup(data, tbl)) {
2019 2 : dbsync_set_error(context,
2020 : "cloudsync_clear_filter: table '%s' is not configured for sync. "
2021 1 : "Call SELECT cloudsync_init('%s') first.", tbl, tbl);
2022 1 : return;
2023 : }
2024 :
2025 : // Remove filter from table settings (set to NULL/empty)
2026 4 : dbutils_table_settings_set_key_value(data, tbl, "*", "filter", NULL);
2027 :
2028 : // Read current algo
2029 4 : table_algo algo = dbutils_table_settings_get_algo(data, tbl);
2030 4 : if (algo == table_algo_none) algo = table_algo_crdt_cls;
2031 :
2032 : // Drop and recreate triggers without filter
2033 4 : database_delete_triggers(data, tbl);
2034 4 : int rc = database_create_triggers(data, tbl, algo, NULL);
2035 4 : if (rc != DBRES_OK) {
2036 0 : dbsync_set_error(context, "cloudsync_clear_filter: error recreating triggers");
2037 0 : sqlite3_result_error_code(context, rc);
2038 0 : return;
2039 : }
2040 :
2041 : // Clean and refill metatable without filter (all rows)
2042 4 : rc = cloudsync_reset_metatable(data, tbl);
2043 4 : if (rc != DBRES_OK) {
2044 0 : dbsync_set_error(context, "cloudsync_clear_filter: error resetting metatable");
2045 0 : sqlite3_result_error_code(context, rc);
2046 0 : return;
2047 : }
2048 :
2049 4 : sqlite3_result_int(context, 1);
2050 5 : }
2051 :
2052 772 : int dbsync_register_functions (sqlite3 *db, char **pzErrMsg) {
2053 772 : int rc = SQLITE_OK;
2054 :
2055 : // there's no built-in way to verify if sqlite3_cloudsync_init has already been called
2056 : // for this specific database connection, we use a workaround: we attempt to retrieve the
2057 : // cloudsync_version and check for an error, an error indicates that initialization has not been performed
2058 772 : if (sqlite3_exec(db, "SELECT cloudsync_version();", NULL, NULL, NULL) == SQLITE_OK) return SQLITE_OK;
2059 :
2060 : // init memory debugger (NOOP in production)
2061 : cloudsync_memory_init(1);
2062 :
2063 : // set fractional-indexing allocator to use cloudsync memory
2064 770 : block_init_allocator();
2065 :
2066 : // init context
2067 770 : void *ctx = cloudsync_context_create(db);
2068 770 : if (!ctx) {
2069 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("Not enought memory to create a database context");
2070 0 : return SQLITE_NOMEM;
2071 : }
2072 :
2073 : // register functions
2074 :
2075 : // PUBLIC functions
2076 770 : rc = dbsync_register_pure_function(db, "cloudsync_version", dbsync_version, 0, pzErrMsg, ctx, cloudsync_context_free);
2077 770 : if (rc != SQLITE_OK) return rc;
2078 :
2079 770 : rc = dbsync_register_function(db, "cloudsync_init", dbsync_init1, 1, pzErrMsg, ctx, NULL);
2080 770 : if (rc != SQLITE_OK) return rc;
2081 :
2082 770 : rc = dbsync_register_function(db, "cloudsync_init", dbsync_init2, 2, pzErrMsg, ctx, NULL);
2083 770 : if (rc != SQLITE_OK) return rc;
2084 :
2085 770 : rc = dbsync_register_function(db, "cloudsync_init", dbsync_init3, 3, pzErrMsg, ctx, NULL);
2086 770 : if (rc != SQLITE_OK) return rc;
2087 :
2088 770 : rc = dbsync_register_function(db, "cloudsync_enable", dbsync_enable, 1, pzErrMsg, ctx, NULL);
2089 770 : if (rc != SQLITE_OK) return rc;
2090 :
2091 770 : rc = dbsync_register_function(db, "cloudsync_disable", dbsync_disable, 1, pzErrMsg, ctx, NULL);
2092 770 : if (rc != SQLITE_OK) return rc;
2093 :
2094 770 : rc = dbsync_register_function(db, "cloudsync_is_enabled", dbsync_is_enabled, 1, pzErrMsg, ctx, NULL);
2095 770 : if (rc != SQLITE_OK) return rc;
2096 :
2097 770 : rc = dbsync_register_function(db, "cloudsync_cleanup", dbsync_cleanup, 1, pzErrMsg, ctx, NULL);
2098 770 : if (rc != SQLITE_OK) return rc;
2099 :
2100 770 : rc = dbsync_register_function(db, "cloudsync_terminate", dbsync_terminate, 0, pzErrMsg, ctx, NULL);
2101 770 : if (rc != SQLITE_OK) return rc;
2102 :
2103 770 : rc = dbsync_register_function(db, "cloudsync_set", dbsync_set, 2, pzErrMsg, ctx, NULL);
2104 770 : if (rc != SQLITE_OK) return rc;
2105 :
2106 770 : rc = dbsync_register_function(db, "cloudsync_set_table", dbsync_set_table, 3, pzErrMsg, ctx, NULL);
2107 770 : if (rc != SQLITE_OK) return rc;
2108 :
2109 770 : rc = dbsync_register_function(db, "cloudsync_set_filter", dbsync_set_filter, 2, pzErrMsg, ctx, NULL);
2110 770 : if (rc != SQLITE_OK) return rc;
2111 :
2112 770 : rc = dbsync_register_function(db, "cloudsync_clear_filter", dbsync_clear_filter, 1, pzErrMsg, ctx, NULL);
2113 770 : if (rc != SQLITE_OK) return rc;
2114 :
2115 770 : rc = dbsync_register_function(db, "cloudsync_set_schema", dbsync_set_schema, 1, pzErrMsg, ctx, NULL);
2116 770 : if (rc != SQLITE_OK) return rc;
2117 :
2118 770 : rc = dbsync_register_function(db, "cloudsync_schema", dbsync_schema, 0, pzErrMsg, ctx, NULL);
2119 770 : if (rc != SQLITE_OK) return rc;
2120 :
2121 770 : rc = dbsync_register_function(db, "cloudsync_table_schema", dbsync_table_schema, 1, pzErrMsg, ctx, NULL);
2122 770 : if (rc != SQLITE_OK) return rc;
2123 :
2124 770 : rc = dbsync_register_function(db, "cloudsync_set_column", dbsync_set_column, 4, pzErrMsg, ctx, NULL);
2125 770 : if (rc != SQLITE_OK) return rc;
2126 :
2127 770 : rc = dbsync_register_function(db, "cloudsync_siteid", dbsync_siteid, 0, pzErrMsg, ctx, NULL);
2128 770 : if (rc != SQLITE_OK) return rc;
2129 :
2130 770 : rc = dbsync_register_function(db, "cloudsync_db_version", dbsync_db_version, 0, pzErrMsg, ctx, NULL);
2131 770 : if (rc != SQLITE_OK) return rc;
2132 :
2133 770 : rc = dbsync_register_function(db, "cloudsync_db_version_next", dbsync_db_version_next, 0, pzErrMsg, ctx, NULL);
2134 770 : if (rc != SQLITE_OK) return rc;
2135 :
2136 770 : rc = dbsync_register_function(db, "cloudsync_db_version_next", dbsync_db_version_next, 1, pzErrMsg, ctx, NULL);
2137 770 : if (rc != SQLITE_OK) return rc;
2138 :
2139 770 : rc = dbsync_register_function(db, "cloudsync_begin_alter", dbsync_begin_alter, 1, pzErrMsg, ctx, NULL);
2140 770 : if (rc != SQLITE_OK) return rc;
2141 :
2142 770 : rc = dbsync_register_function(db, "cloudsync_commit_alter", dbsync_commit_alter, 1, pzErrMsg, ctx, NULL);
2143 770 : if (rc != SQLITE_OK) return rc;
2144 :
2145 770 : rc = dbsync_register_function(db, "cloudsync_uuid", dbsync_uuid, 0, pzErrMsg, ctx, NULL);
2146 770 : if (rc != SQLITE_OK) return rc;
2147 :
2148 770 : rc = dbsync_register_function(db, "cloudsync_uuid_text", dbsync_uuid_text, 1, pzErrMsg, ctx, NULL);
2149 770 : if (rc != SQLITE_OK) return rc;
2150 770 : rc = dbsync_register_function(db, "cloudsync_uuid_text", dbsync_uuid_text, 2, pzErrMsg, ctx, NULL);
2151 770 : if (rc != SQLITE_OK) return rc;
2152 770 : rc = dbsync_register_function(db, "cloudsync_uuid_blob", dbsync_uuid_blob, 1, pzErrMsg, ctx, NULL);
2153 770 : if (rc != SQLITE_OK) return rc;
2154 :
2155 : // PAYLOAD
2156 770 : rc = dbsync_register_aggregate(db, "cloudsync_payload_encode", dbsync_payload_encode_step, dbsync_payload_encode_final, -1, pzErrMsg, ctx, NULL);
2157 770 : if (rc != SQLITE_OK) return rc;
2158 :
2159 : // alias
2160 770 : rc = dbsync_register_function(db, "cloudsync_payload_decode", dbsync_payload_decode, -1, pzErrMsg, ctx, NULL);
2161 770 : if (rc != SQLITE_OK) return rc;
2162 770 : rc = dbsync_register_function(db, "cloudsync_payload_apply", dbsync_payload_decode, -1, pzErrMsg, ctx, NULL);
2163 770 : if (rc != SQLITE_OK) return rc;
2164 :
2165 770 : rc = sqlite3_create_module(db, "cloudsync_payload_chunks", &cloudsync_payload_chunks_module, (void *)ctx);
2166 770 : if (rc != SQLITE_OK) return rc;
2167 :
2168 770 : rc = dbsync_register_function(db, "cloudsync_payload_blob_checked", dbsync_payload_blob_checked, 5, pzErrMsg, ctx, NULL);
2169 770 : if (rc != SQLITE_OK) return rc;
2170 :
2171 : #ifdef CLOUDSYNC_DESKTOP_OS
2172 770 : rc = dbsync_register_function(db, "cloudsync_payload_save", dbsync_payload_save, 1, pzErrMsg, ctx, NULL);
2173 770 : if (rc != SQLITE_OK) return rc;
2174 :
2175 770 : rc = dbsync_register_function(db, "cloudsync_payload_load", dbsync_payload_load, 1, pzErrMsg, ctx, NULL);
2176 770 : if (rc != SQLITE_OK) return rc;
2177 : #endif
2178 :
2179 : // PRIVATE functions (used inside triggers — require SQLITE_INNOCUOUS)
2180 770 : rc = dbsync_register_trigger_function(db, "cloudsync_is_sync", dbsync_is_sync, 1, pzErrMsg, ctx, NULL);
2181 770 : if (rc != SQLITE_OK) return rc;
2182 :
2183 770 : rc = dbsync_register_trigger_function(db, "cloudsync_insert", dbsync_insert, -1, pzErrMsg, ctx, NULL);
2184 770 : if (rc != SQLITE_OK) return rc;
2185 :
2186 770 : rc = dbsync_register_trigger_aggregate(db, "cloudsync_update", dbsync_update_step, dbsync_update_final, 3, pzErrMsg, ctx, NULL);
2187 770 : if (rc != SQLITE_OK) return rc;
2188 :
2189 770 : rc = dbsync_register_trigger_function(db, "cloudsync_delete", dbsync_delete, -1, pzErrMsg, ctx, NULL);
2190 770 : if (rc != SQLITE_OK) return rc;
2191 :
2192 770 : rc = dbsync_register_function(db, "cloudsync_col_value", dbsync_col_value, 3, pzErrMsg, ctx, NULL);
2193 770 : if (rc != SQLITE_OK) return rc;
2194 :
2195 770 : rc = dbsync_register_pure_function(db, "cloudsync_pk_encode", dbsync_pk_encode, -1, pzErrMsg, ctx, NULL);
2196 770 : if (rc != SQLITE_OK) return rc;
2197 :
2198 770 : rc = dbsync_register_pure_function(db, "cloudsync_pk_decode", dbsync_pk_decode, 2, pzErrMsg, ctx, NULL);
2199 770 : if (rc != SQLITE_OK) return rc;
2200 :
2201 770 : rc = dbsync_register_function(db, "cloudsync_seq", dbsync_seq, 0, pzErrMsg, ctx, NULL);
2202 770 : if (rc != SQLITE_OK) return rc;
2203 :
2204 770 : rc = dbsync_register_function(db, "cloudsync_text_materialize", dbsync_text_materialize, -1, pzErrMsg, ctx, NULL);
2205 770 : if (rc != SQLITE_OK) return rc;
2206 :
2207 : // NETWORK LAYER
2208 : #ifndef CLOUDSYNC_OMIT_NETWORK
2209 20 : rc = cloudsync_network_register(db, pzErrMsg, ctx);
2210 20 : if (rc != SQLITE_OK) return rc;
2211 : #endif
2212 :
2213 770 : cloudsync_context *data = (cloudsync_context *)ctx;
2214 770 : sqlite3_commit_hook(db, cloudsync_commit_hook, ctx);
2215 770 : sqlite3_rollback_hook(db, cloudsync_rollback_hook, ctx);
2216 :
2217 : // register eponymous only changes virtual table
2218 770 : rc = cloudsync_vtab_register_changes (db, data);
2219 770 : if (rc != SQLITE_OK) {
2220 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("Error creating changes virtual table: %s", sqlite3_errmsg(db));
2221 0 : return rc;
2222 : }
2223 :
2224 : // load config, if exists
2225 770 : if (cloudsync_config_exists(data)) {
2226 6 : if (cloudsync_context_init(data) == NULL) {
2227 : // Do not free ctx here: it is already owned by the cloudsync_version
2228 : // function (registered above with cloudsync_context_free as its
2229 : // destructor). SQLite will release it when the connection is closed.
2230 : // Freeing it manually would cause a double-free on sqlite3_close.
2231 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("An error occurred while trying to initialize context");
2232 0 : return SQLITE_ERROR;
2233 : }
2234 :
2235 : // update schema hash if upgrading from an older version
2236 6 : if (dbutils_settings_check_version(data, NULL) != 0) {
2237 0 : cloudsync_update_schema_hash(data);
2238 0 : }
2239 :
2240 : // make sure to update internal version to current version
2241 6 : dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_LIBVERSION, CLOUDSYNC_VERSION);
2242 6 : }
2243 :
2244 770 : return SQLITE_OK;
2245 772 : }
2246 :
2247 : // MARK: - Main Entrypoint -
2248 :
2249 772 : APIEXPORT int sqlite3_cloudsync_init (sqlite3 *db, char **pzErrMsg, const sqlite3_api_routines *pApi) {
2250 : DEBUG_FUNCTION("sqlite3_cloudsync_init");
2251 :
2252 : #ifndef SQLITE_CORE
2253 : SQLITE_EXTENSION_INIT2(pApi);
2254 : #endif
2255 :
2256 772 : return dbsync_register_functions(db, pzErrMsg);
2257 : }
|
