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 3 : void dbsync_db_version (sqlite3_context *context, int argc, sqlite3_value **argv) {
91 : DEBUG_FUNCTION("cloudsync_db_version");
92 3 : UNUSED_PARAMETER(argc);
93 3 : UNUSED_PARAMETER(argv);
94 :
95 : // retrieve context
96 3 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
97 :
98 3 : int rc = cloudsync_dbversion_check_uptodate(data);
99 3 : 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 2 : sqlite3_result_int64(context, cloudsync_dbversion(data));
118 3 : }
119 :
120 32203 : 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 32203 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
125 :
126 32203 : sqlite3_int64 merging_version = (argc == 1) ? database_value_int(argv[0]) : CLOUDSYNC_VALUE_NOTSET;
127 32203 : sqlite3_int64 value = cloudsync_dbversion_next(data, merging_version);
128 32203 : 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 32202 : sqlite3_result_int64(context, value);
140 32203 : }
141 :
142 58 : void dbsync_seq (sqlite3_context *context, int argc, sqlite3_value **argv) {
143 : DEBUG_FUNCTION("cloudsync_seq");
144 :
145 : // retrieve context
146 58 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
147 58 : sqlite3_result_int(context, cloudsync_bumpseq(data));
148 58 : }
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 17 : void dbsync_set (sqlite3_context *context, int argc, sqlite3_value **argv) {
191 : DEBUG_FUNCTION("cloudsync_set");
192 :
193 : // sanity check parameters
194 17 : const char *key = (const char *)database_value_text(argv[0]);
195 17 : const char *value = (const char *)database_value_text(argv[1]);
196 :
197 : // silently fails
198 17 : if (key == NULL) return;
199 :
200 17 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
201 17 : dbutils_settings_set_key_value(data, key, value);
202 17 : }
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 20701 : void dbsync_is_sync (sqlite3_context *context, int argc, sqlite3_value **argv) {
274 : DEBUG_FUNCTION("cloudsync_is_sync");
275 :
276 20701 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
277 20701 : if (cloudsync_insync(data)) {
278 14378 : sqlite3_result_int(context, 1);
279 14378 : return;
280 : }
281 :
282 6323 : const char *table_name = (const char *)database_value_text(argv[0]);
283 6323 : cloudsync_table_context *table = table_lookup(data, table_name);
284 6323 : sqlite3_result_int(context, (table) ? (table_enabled(table) == 0) : 0);
285 20701 : }
286 :
287 151336 : 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 151336 : const char *col_name = (const char *)database_value_text(argv[1]);
296 151336 : 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 151336 : if (strcmp(col_name, CLOUDSYNC_TOMBSTONE_VALUE) == 0) {
303 4796 : sqlite3_result_null(context);
304 4796 : return;
305 : }
306 :
307 : // lookup table
308 146540 : const char *table_name = (const char *)database_value_text(argv[0]);
309 146540 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
310 146540 : cloudsync_table_context *table = table_lookup(data, table_name);
311 146540 : 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 146540 : 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 134305 : sqlite3_stmt *vm = table_column_lookup(table, col_name, false, NULL);
342 134305 : 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 134305 : 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 134305 : if (rc < 0) goto cleanup;
350 :
351 : // execute vm
352 134305 : rc = databasevm_step(vm);
353 268610 : if (rc == SQLITE_DONE) {
354 0 : rc = SQLITE_OK;
355 0 : sqlite3_result_text(context, CLOUDSYNC_RLS_RESTRICTED_VALUE, -1, SQLITE_STATIC);
356 134305 : } else if (rc == SQLITE_ROW) {
357 : // store value result
358 134305 : rc = SQLITE_OK;
359 134305 : sqlite3_result_value(context, database_column_value(vm, 0));
360 134305 : }
361 :
362 : cleanup:
363 134305 : if (rc != SQLITE_OK) {
364 0 : sqlite3_result_error(context, database_errmsg(data), -1);
365 0 : }
366 134305 : databasevm_reset(vm);
367 151336 : }
368 :
369 10699 : void dbsync_pk_encode (sqlite3_context *context, int argc, sqlite3_value **argv) {
370 10699 : size_t bsize = 0;
371 10699 : char *buffer = pk_encode_prikey((dbvalue_t **)argv, argc, NULL, &bsize);
372 10699 : if (!buffer || buffer == PRIKEY_NULL_CONSTRAINT_ERROR) {
373 1 : sqlite3_result_null(context);
374 1 : return;
375 : }
376 10698 : sqlite3_result_blob(context, (const void *)buffer, (int)bsize, SQLITE_TRANSIENT);
377 10698 : cloudsync_memory_free(buffer);
378 10699 : }
379 :
380 21 : int dbsync_pk_decode_set_result_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
381 21 : cloudsync_pk_decode_context *decode_context = (cloudsync_pk_decode_context *)xdata;
382 : // decode_context->index is 1 based
383 : // index is 0 based
384 21 : if (decode_context->index != index+1) return SQLITE_OK;
385 :
386 9 : int rc = 0;
387 9 : sqlite3_context *context = decode_context->context;
388 9 : 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 1 : sqlite3_result_text(context, pval, (int)ival, SQLITE_TRANSIENT);
403 1 : break;
404 :
405 : case SQLITE_BLOB:
406 1 : sqlite3_result_blob(context, pval, (int)ival, SQLITE_TRANSIENT);
407 1 : break;
408 : }
409 :
410 9 : return rc;
411 21 : }
412 :
413 :
414 9 : void dbsync_pk_decode (sqlite3_context *context, int argc, sqlite3_value **argv) {
415 9 : const char *pk = (const char *)database_value_blob(argv[0]);
416 9 : int pk_len = database_value_bytes(argv[0]);
417 9 : int i = (int)database_value_int(argv[1]);
418 :
419 9 : cloudsync_pk_decode_context xdata = {.context = context, .index = i};
420 9 : pk_decode_prikey((char *)pk, (size_t)pk_len, dbsync_pk_decode_set_result_callback, &xdata);
421 9 : }
422 :
423 : // MARK: -
424 :
425 5560 : 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 5560 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
442 5560 : cloudsync_reset_error(data);
443 :
444 : // lookup table
445 5560 : const char *table_name = (const char *)database_value_text(argv[0]);
446 5560 : cloudsync_table_context *table = table_lookup(data, table_name);
447 5560 : 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 5560 : size_t pklen = sizeof(buffer);
455 5560 : char *pk = pk_encode_prikey((dbvalue_t **)&argv[1], table_count_pks(table), buffer, &pklen);
456 5560 : if (!pk) {
457 0 : sqlite3_result_error(context, "Not enough memory to encode the primary key(s).", -1);
458 0 : return;
459 : }
460 5560 : 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 5559 : 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 5559 : bool pk_exists = table_pk_exists(table, pk, pklen);
471 5559 : int rc = SQLITE_OK;
472 :
473 5559 : 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 5559 : } 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 19017 : for (int i=0; i<table_count_cols(table); ++i) {
485 13461 : 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 12759 : rc = local_mark_insert_or_update_meta(table, pk, pklen, table_colname(table, i), db_version, cloudsync_bumpseq(data));
491 12759 : if (rc != SQLITE_OK) goto cleanup;
492 : }
493 19014 : }
494 :
495 : cleanup:
496 5559 : if (rc != SQLITE_OK) dbsync_result_trigger_error(context, data, rc);
497 : // free memory if the primary key was dynamically allocated
498 5559 : if (pk != buffer) cloudsync_memory_free(pk);
499 5560 : }
500 :
501 201 : 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 201 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
507 201 : cloudsync_reset_error(data);
508 :
509 : // lookup table
510 201 : const char *table_name = (const char *)database_value_text(argv[0]);
511 201 : cloudsync_table_context *table = table_lookup(data, table_name);
512 201 : 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 201 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
519 201 : int rc = SQLITE_OK;
520 :
521 : // encode the primary key values into a buffer
522 : char buffer[1024];
523 201 : size_t pklen = sizeof(buffer);
524 201 : char *pk = pk_encode_prikey((dbvalue_t **)&argv[1], table_count_pks(table), buffer, &pklen);
525 201 : if (!pk) {
526 0 : sqlite3_result_error(context, "Not enough memory to encode the primary key(s).", -1);
527 0 : return;
528 : }
529 :
530 201 : 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 201 : rc = local_mark_delete_meta(table, pk, pklen, db_version, cloudsync_bumpseq(data));
537 201 : if (rc != SQLITE_OK) goto cleanup;
538 :
539 : // remove any metadata related to the old rows associated with this primary key
540 201 : rc = local_drop_meta(table, pk, pklen);
541 201 : if (rc != SQLITE_OK) goto cleanup;
542 :
543 : cleanup:
544 201 : if (rc != SQLITE_OK) dbsync_result_trigger_error(context, data, rc);
545 : // free memory if the primary key was dynamically allocated
546 201 : if (pk != buffer) cloudsync_memory_free(pk);
547 201 : }
548 :
549 : // MARK: -
550 :
551 608 : void dbsync_update_payload_free (cloudsync_update_payload *payload) {
552 3299 : for (int i=0; i<payload->count; i++) {
553 2691 : database_value_free(payload->new_values[i]);
554 2691 : database_value_free(payload->old_values[i]);
555 2691 : }
556 608 : cloudsync_memory_free(payload->new_values);
557 608 : cloudsync_memory_free(payload->old_values);
558 608 : database_value_free(payload->table_name);
559 608 : payload->new_values = NULL;
560 608 : payload->old_values = NULL;
561 608 : payload->table_name = NULL;
562 608 : payload->count = 0;
563 608 : payload->capacity = 0;
564 608 : }
565 :
566 2691 : int dbsync_update_payload_append (cloudsync_update_payload *payload, sqlite3_value *v1, sqlite3_value *v2, sqlite3_value *v3) {
567 2691 : if (payload->count >= payload->capacity) {
568 611 : int newcap = payload->capacity ? payload->capacity * 2 : 128;
569 :
570 611 : sqlite3_value **new_values_2 = (sqlite3_value **)cloudsync_memory_realloc(payload->new_values, newcap * sizeof(*new_values_2));
571 611 : if (!new_values_2) return SQLITE_NOMEM;
572 :
573 611 : sqlite3_value **old_values_2 = (sqlite3_value **)cloudsync_memory_realloc(payload->old_values, newcap * sizeof(*old_values_2));
574 611 : 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 611 : payload->new_values = new_values_2;
582 611 : payload->old_values = old_values_2;
583 611 : payload->capacity = newcap;
584 611 : }
585 :
586 2691 : int index = payload->count;
587 2691 : if (payload->table_name == NULL) payload->table_name = database_value_dup(v1);
588 2083 : else if (dbutils_value_compare(payload->table_name, v1) != 0) return SQLITE_NOMEM;
589 :
590 2691 : payload->new_values[index] = database_value_dup(v2);
591 2691 : payload->old_values[index] = database_value_dup(v3);
592 :
593 : // sanity check memory allocations before committing count
594 2691 : bool v1_can_be_null = (database_value_type(v1) == SQLITE_NULL);
595 2691 : bool v2_can_be_null = (database_value_type(v2) == SQLITE_NULL);
596 2691 : bool v3_can_be_null = (database_value_type(v3) == SQLITE_NULL);
597 :
598 2691 : bool oom = false;
599 2691 : if ((payload->table_name == NULL) && (!v1_can_be_null)) oom = true;
600 2691 : if ((payload->new_values[index] == NULL) && (!v2_can_be_null)) oom = true;
601 2691 : if ((payload->old_values[index] == NULL) && (!v3_can_be_null)) oom = true;
602 :
603 2691 : 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 2691 : payload->count++;
611 2691 : return SQLITE_OK;
612 2691 : }
613 :
614 2691 : 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 2691 : cloudsync_update_payload *payload = (cloudsync_update_payload *)sqlite3_aggregate_context(context, sizeof(cloudsync_update_payload));
621 2691 : if (!payload) {sqlite3_result_error_nomem(context); return;}
622 :
623 2691 : if (dbsync_update_payload_append(payload, argv[0], argv[1], argv[2]) != SQLITE_OK) {
624 0 : sqlite3_result_error_nomem(context);
625 0 : }
626 2691 : }
627 :
628 608 : void dbsync_update_final (sqlite3_context *context) {
629 608 : cloudsync_update_payload *payload = (cloudsync_update_payload *)sqlite3_aggregate_context(context, sizeof(cloudsync_update_payload));
630 608 : if (!payload || payload->count == 0) return;
631 :
632 : // retrieve context
633 608 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
634 608 : cloudsync_reset_error(data);
635 :
636 : // lookup table
637 608 : const char *table_name = (const char *)database_value_text(payload->table_name);
638 608 : cloudsync_table_context *table = table_lookup(data, table_name);
639 608 : 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 608 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
647 608 : int rc = SQLITE_OK;
648 :
649 : // Check if the primary key(s) have changed
650 608 : bool prikey_changed = false;
651 1496 : for (int i=0; i<table_count_pks(table); ++i) {
652 917 : if (dbutils_value_compare(payload->old_values[i], payload->new_values[i]) != 0) {
653 29 : prikey_changed = true;
654 29 : break;
655 : }
656 888 : }
657 :
658 : // encode the NEW primary key values into a buffer (used later for indexing)
659 : char buffer[1024];
660 : char buffer2[1024];
661 608 : size_t pklen = sizeof(buffer);
662 608 : size_t oldpklen = sizeof(buffer2);
663 608 : char *oldpk = NULL;
664 :
665 608 : char *pk = pk_encode_prikey((dbvalue_t **)payload->new_values, table_count_pks(table), buffer, &pklen);
666 608 : 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 608 : 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 608 : 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 29 : oldpk = pk_encode_prikey((dbvalue_t **)payload->old_values, table_count_pks(table), buffer2, &oldpklen);
684 29 : 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 29 : rc = local_mark_delete_meta(table, oldpk, oldpklen, db_version, cloudsync_bumpseq(data));
694 29 : 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 29 : rc = local_update_move_meta(table, pk, pklen, oldpk, oldpklen, db_version);
700 29 : if (rc != SQLITE_OK) goto cleanup;
701 :
702 : // mark a new sentinel row with the new primary key in the metadata
703 29 : rc = local_mark_insert_sentinel_meta(table, pk, pklen, db_version, cloudsync_bumpseq(data));
704 29 : if (rc != SQLITE_OK) goto cleanup;
705 :
706 : // free memory if the OLD primary key was dynamically allocated
707 29 : if (oldpk != buffer2) cloudsync_memory_free(oldpk);
708 29 : oldpk = NULL;
709 29 : }
710 :
711 : // compare NEW and OLD values (excluding primary keys) to handle column updates
712 2353 : for (int i=0; i<table_count_cols(table); i++) {
713 1746 : int col_index = table_count_pks(table) + i; // Regular columns start after primary keys
714 :
715 1746 : if (dbutils_value_compare(payload->old_values[col_index], payload->new_values[col_index]) != 0) {
716 927 : 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 672 : rc = local_mark_insert_or_update_meta(table, pk, pklen, table_colname(table, i), db_version, cloudsync_bumpseq(data));
723 672 : if (rc != SQLITE_OK) goto cleanup;
724 : }
725 926 : }
726 2352 : }
727 :
728 : cleanup:
729 608 : if (rc != SQLITE_OK) dbsync_result_trigger_error(context, data, rc);
730 608 : if (pk != buffer) cloudsync_memory_free(pk);
731 608 : if (oldpk && (oldpk != buffer2)) cloudsync_memory_free(oldpk);
732 :
733 608 : dbsync_update_payload_free(payload);
734 608 : }
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 761 : void dbsync_terminate (sqlite3_context *context, int argc, sqlite3_value **argv) {
787 : DEBUG_FUNCTION("cloudsync_terminate");
788 :
789 761 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
790 761 : int rc = cloudsync_terminate(data);
791 761 : sqlite3_result_int(context, rc);
792 761 : }
793 :
794 : // MARK: -
795 :
796 790 : void dbsync_init (sqlite3_context *context, const char *table, const char *algo, CLOUDSYNC_INIT_FLAG init_flags) {
797 790 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
798 :
799 790 : int rc = database_begin_savepoint(data, "cloudsync_init");
800 790 : 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 790 : rc = cloudsync_init_table(data, table, algo, init_flags);
807 790 : if (rc == SQLITE_OK) {
808 785 : rc = database_commit_savepoint(data, "cloudsync_init");
809 785 : 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 785 : } 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 785 : cloudsync_update_schema_hash(data);
822 :
823 : // returns site_id as TEXT
824 : char buffer[UUID_STR_MAXLEN];
825 785 : cloudsync_uuid_v7_stringify(cloudsync_siteid(data), buffer, false);
826 785 : sqlite3_result_text(context, buffer, -1, SQLITE_TRANSIENT);
827 790 : }
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 658 : void dbsync_init1 (sqlite3_context *context, int argc, sqlite3_value **argv) {
847 : DEBUG_FUNCTION("cloudsync_init1");
848 :
849 658 : const char *table = (const char *)database_value_text(argv[0]);
850 658 : dbsync_init(context, table, NULL, CLOUDSYNC_INIT_FLAG_NONE);
851 658 : }
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 56716 : void dbsync_payload_encode_step (sqlite3_context *context, int argc, sqlite3_value **argv) {
906 : // allocate/get the session context
907 56716 : cloudsync_payload_context *payload = (cloudsync_payload_context *)sqlite3_aggregate_context(context, (int)cloudsync_payload_context_size(NULL));
908 56716 : 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 56716 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
916 :
917 56716 : int rc = cloudsync_payload_encode_step(payload, data, argc, (dbvalue_t **)argv);
918 56716 : 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 56716 : }
923 :
924 1808 : void dbsync_payload_encode_final (sqlite3_context *context) {
925 : // get the session context
926 1808 : cloudsync_payload_context *payload = (cloudsync_payload_context *)sqlite3_aggregate_context(context, (int)cloudsync_payload_context_size(NULL));
927 1808 : 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 1808 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
935 :
936 1808 : int rc = cloudsync_payload_encode_final(payload, data);
937 1808 : 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 1808 : int64_t blob_size = 0;
945 1808 : char *blob = cloudsync_payload_blob (payload, &blob_size, NULL);
946 1808 : if (!blob) {
947 8 : sqlite3_result_null(context);
948 8 : } else {
949 1800 : sqlite3_result_blob64(context, blob, blob_size, SQLITE_TRANSIENT);
950 1800 : 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 1808 : }
956 :
957 1882 : 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 1882 : 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 1882 : int blen = database_value_bytes(argv[0]);
970 1882 : size_t header_size = 0;
971 1882 : cloudsync_payload_context_size(&header_size);
972 1882 : 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 1879 : 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 1879 : int nrows = 0;
988 1879 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
989 1879 : int rc = cloudsync_payload_apply(data, payload, blen, &nrows, CLOUDSYNC_CHECKPOINT_LAST_APPLIED, 0);
990 1879 : 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 1607 : sqlite3_result_int(context, nrows);
998 1882 : }
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 : } cloudsync_payload_chunks_cursor;
1039 :
1040 12 : static int payload_chunks_connect(sqlite3 *db, void *aux, int argc, const char *const *argv, sqlite3_vtab **vtab, char **err) {
1041 12 : UNUSED_PARAMETER(argc); UNUSED_PARAMETER(argv); UNUSED_PARAMETER(err);
1042 12 : int rc = sqlite3_declare_vtab(db,
1043 : "CREATE TABLE x(payload BLOB, chunk_index INTEGER, payload_size INTEGER, rows INTEGER, "
1044 : "db_version_min INTEGER, db_version_max INTEGER, watermark_db_version INTEGER, "
1045 : "since_db_version HIDDEN, site_id HIDDEN, until_db_version HIDDEN, exclude_filter_site_id HIDDEN, "
1046 : // Positional-cursor outputs (cols 11..14): the resume point after the
1047 : // emitted chunk, plus a final-chunk flag. A stateless /check passes these
1048 : // back as the resume_* inputs (cols 15..17) to continue the drain without
1049 : // a spool table — O(1) seek per chunk instead of replaying from since.
1050 : "next_db_version INTEGER, next_seq INTEGER, next_frag_offset INTEGER, is_final INTEGER, "
1051 : "resume_db_version HIDDEN, resume_seq HIDDEN, resume_frag_offset HIDDEN)");
1052 12 : if (rc != SQLITE_OK) return rc;
1053 12 : cloudsync_payload_chunks_vtab *p = sqlite3_malloc64(sizeof(*p));
1054 12 : if (!p) return SQLITE_NOMEM;
1055 12 : memset(p, 0, sizeof(*p));
1056 12 : p->db = db;
1057 12 : p->data = (cloudsync_context *)aux;
1058 12 : *vtab = (sqlite3_vtab *)p;
1059 12 : return SQLITE_OK;
1060 12 : }
1061 :
1062 12 : static int payload_chunks_disconnect(sqlite3_vtab *vtab) {
1063 12 : sqlite3_free(vtab);
1064 12 : return SQLITE_OK;
1065 : }
1066 :
1067 29 : static int payload_chunks_open(sqlite3_vtab *vtab, sqlite3_vtab_cursor **cursor) {
1068 29 : cloudsync_payload_chunks_cursor *c = cloudsync_memory_zeroalloc(sizeof(*c));
1069 29 : if (!c) return SQLITE_NOMEM;
1070 29 : c->vtab = (cloudsync_payload_chunks_vtab *)vtab;
1071 29 : *cursor = (sqlite3_vtab_cursor *)c;
1072 29 : return SQLITE_OK;
1073 29 : }
1074 :
1075 29 : static int payload_chunks_close(sqlite3_vtab_cursor *cursor) {
1076 29 : cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
1077 29 : if (c->src) sqlite3_finalize(c->src);
1078 29 : if (c->payload) cloudsync_memory_free(c->payload);
1079 29 : cloudsync_memory_free(c);
1080 29 : return SQLITE_OK;
1081 : }
1082 :
1083 20 : static int payload_chunks_best_index(sqlite3_vtab *vtab, sqlite3_index_info *idxinfo) {
1084 20 : UNUSED_PARAMETER(vtab);
1085 : // Assign argvIndex in a canonical hidden-column order so xFilter can read argv
1086 : // in a fixed order regardless of how SQLite presents constraints. idxNum bit k
1087 : // is set when handled_cols[k] is bound; xFilter reads argv in this same order.
1088 : // bit0=since_db_version(7) bit1=site_id(8) bit2=until_db_version(9)
1089 : // bit3=exclude_filter_site_id(10) bit4=resume_db_version(15)
1090 : // bit5=resume_seq(16) bit6=resume_frag_offset(17)
1091 : static const int handled_cols[] = {7, 8, 9, 10, 15, 16, 17};
1092 20 : int argv_index = 1;
1093 20 : int idxnum = 0;
1094 160 : for (size_t k = 0; k < sizeof(handled_cols) / sizeof(handled_cols[0]); ++k) {
1095 140 : int col = handled_cols[k];
1096 262 : for (int i = 0; i < idxinfo->nConstraint; ++i) {
1097 144 : struct sqlite3_index_constraint *cn = &idxinfo->aConstraint[i];
1098 144 : if (!cn->usable || cn->op != SQLITE_INDEX_CONSTRAINT_EQ || cn->iColumn != col) continue;
1099 22 : idxinfo->aConstraintUsage[i].argvIndex = argv_index++;
1100 22 : idxinfo->aConstraintUsage[i].omit = 1;
1101 22 : idxnum |= (1 << k);
1102 22 : break; // at most one constraint consumed per hidden column
1103 : }
1104 140 : }
1105 20 : idxinfo->idxNum = idxnum;
1106 20 : idxinfo->estimatedCost = 10.0;
1107 20 : idxinfo->estimatedRows = 10;
1108 20 : return SQLITE_OK;
1109 : }
1110 :
1111 2079 : static int payload_chunks_step_source(cloudsync_payload_chunks_cursor *c) {
1112 2079 : int rc = sqlite3_step(c->src);
1113 2079 : if (rc == SQLITE_ROW) { c->has_row = true; return SQLITE_OK; }
1114 18 : c->has_row = false;
1115 18 : if (rc == SQLITE_DONE) return SQLITE_OK;
1116 : // copy the inner statement's message onto this vtab or SQLite surfaces the
1117 : // error as a bare "SQL logic error"
1118 0 : if (c->vtab->base.zErrMsg) sqlite3_free(c->vtab->base.zErrMsg);
1119 0 : c->vtab->base.zErrMsg = sqlite3_mprintf("%s", sqlite3_errmsg(c->vtab->db));
1120 0 : return rc;
1121 2079 : }
1122 :
1123 21 : static int payload_chunks_plan_fragment(cloudsync_payload_chunks_cursor *c) {
1124 21 : cloudsync_context *data = c->vtab->data;
1125 42 : int target = cloudsync_payload_fragment_data_size(data,
1126 21 : (const char *)sqlite3_column_text(c->src, 0), sqlite3_column_bytes(c->src, 0),
1127 21 : sqlite3_column_blob(c->src, 1), sqlite3_column_bytes(c->src, 1),
1128 21 : (const char *)sqlite3_column_text(c->src, 2), sqlite3_column_bytes(c->src, 2),
1129 21 : sqlite3_column_int64(c->src, 4), sqlite3_column_int64(c->src, 5),
1130 21 : sqlite3_column_blob(c->src, 6), sqlite3_column_bytes(c->src, 6),
1131 21 : sqlite3_column_int64(c->src, 7), sqlite3_column_int64(c->src, 8),
1132 21 : c->frag_total, 0, 1);
1133 21 : if (target <= 0) return SQLITE_TOOBIG;
1134 :
1135 21 : int count = 0;
1136 21 : for (int i = 0; i < CLOUDSYNC_PAYLOAD_FRAGMENT_SIZE_FIXPOINT_ITERATIONS; ++i) {
1137 21 : count = cloudsync_payload_fragment_count(c->frag_total, target);
1138 21 : if (count <= 0) return SQLITE_TOOBIG;
1139 42 : int planned = cloudsync_payload_fragment_data_size(data,
1140 21 : (const char *)sqlite3_column_text(c->src, 0), sqlite3_column_bytes(c->src, 0),
1141 21 : sqlite3_column_blob(c->src, 1), sqlite3_column_bytes(c->src, 1),
1142 21 : (const char *)sqlite3_column_text(c->src, 2), sqlite3_column_bytes(c->src, 2),
1143 21 : sqlite3_column_int64(c->src, 4), sqlite3_column_int64(c->src, 5),
1144 21 : sqlite3_column_blob(c->src, 6), sqlite3_column_bytes(c->src, 6),
1145 21 : sqlite3_column_int64(c->src, 7), sqlite3_column_int64(c->src, 8),
1146 21 : c->frag_total, count - 1, count);
1147 21 : if (planned <= 0) return SQLITE_TOOBIG;
1148 21 : if (planned == target) break;
1149 0 : target = planned;
1150 0 : }
1151 :
1152 21 : c->frag_target = target;
1153 21 : c->frag_count = cloudsync_payload_fragment_count(c->frag_total, target);
1154 21 : if (c->frag_count <= 0) return SQLITE_TOOBIG;
1155 21 : return SQLITE_OK;
1156 21 : }
1157 :
1158 : // Set up fragment state for the current source row (a single value larger than
1159 : // max_chunk_size) so emit_fragment can stream it. start_offset is the byte offset
1160 : // within the encoded value to resume from (0 when first reaching the value;
1161 : // >0 when a positional cursor resumes mid-value). frag_part is derived from the
1162 : // offset so the fragment's part index is consistent whether reached by streaming
1163 : // or by a seek. The plan (frag_target/frag_count) is a deterministic function of
1164 : // the row, so a resumed fragment tiles identically to a streamed one.
1165 21 : static int payload_chunks_begin_fragment(cloudsync_payload_chunks_cursor *c, int64_t start_offset) {
1166 21 : dbvalue_t *col_value = (dbvalue_t *)sqlite3_column_value(c->src, 3);
1167 21 : int type = database_value_type(col_value);
1168 21 : if (type != DBTYPE_TEXT && type != DBTYPE_BLOB) return SQLITE_TOOBIG;
1169 21 : int64_t raw_len = 0;
1170 21 : int header_len = cloudsync_payload_encoded_value_header(col_value, c->value_header, sizeof(c->value_header), &raw_len);
1171 21 : if (header_len <= 0) return SQLITE_ERROR;
1172 21 : c->value_header_len = header_len;
1173 21 : c->value_data = (const char *)database_value_blob(col_value);
1174 21 : c->value_data_len = raw_len;
1175 21 : c->frag_total = header_len + raw_len;
1176 21 : c->frag_offset = start_offset;
1177 21 : int rc = payload_chunks_plan_fragment(c);
1178 21 : if (rc != SQLITE_OK) return rc;
1179 21 : c->frag_part = (c->frag_target > 0) ? (int)(start_offset / c->frag_target) : 0;
1180 21 : c->frag_checksum = cloudsync_payload_encoded_value_checksum(col_value);
1181 21 : c->frag_active = true;
1182 21 : return SQLITE_OK;
1183 21 : }
1184 :
1185 53 : static int payload_chunks_emit_fragment(cloudsync_payload_chunks_cursor *c) {
1186 53 : cloudsync_context *data = c->vtab->data;
1187 53 : if (c->payload) { cloudsync_memory_free(c->payload); c->payload = NULL; }
1188 53 : int64_t remaining = c->frag_total - c->frag_offset;
1189 53 : int frag_len = remaining > c->frag_target ? c->frag_target : (int)remaining;
1190 53 : if (frag_len <= 0) return SQLITE_CORRUPT;
1191 53 : char *frag = cloudsync_memory_alloc((uint64_t)frag_len);
1192 53 : if (!frag) return SQLITE_NOMEM;
1193 53 : int copied = 0;
1194 53 : int64_t off = c->frag_offset;
1195 53 : if (off < c->value_header_len) {
1196 15 : int n = c->value_header_len - (int)off;
1197 15 : if (n > frag_len) n = frag_len;
1198 15 : memcpy(frag, c->value_header + off, (size_t)n);
1199 15 : copied += n;
1200 15 : off += n;
1201 15 : }
1202 53 : if (copied < frag_len) {
1203 53 : int64_t data_off = off - c->value_header_len;
1204 53 : memcpy(frag + copied, c->value_data + data_off, (size_t)(frag_len - copied));
1205 53 : }
1206 :
1207 53 : cloudsync_payload_context *payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
1208 53 : if (!payload) { cloudsync_memory_free(frag); return SQLITE_NOMEM; }
1209 106 : int rc = cloudsync_payload_encode_fragment_step(payload, data,
1210 53 : (const char *)sqlite3_column_text(c->src, 0), sqlite3_column_bytes(c->src, 0),
1211 53 : sqlite3_column_blob(c->src, 1), sqlite3_column_bytes(c->src, 1),
1212 53 : (const char *)sqlite3_column_text(c->src, 2), sqlite3_column_bytes(c->src, 2),
1213 53 : frag, frag_len,
1214 53 : sqlite3_column_int64(c->src, 4), sqlite3_column_int64(c->src, 5),
1215 53 : sqlite3_column_blob(c->src, 6), sqlite3_column_bytes(c->src, 6),
1216 53 : sqlite3_column_int64(c->src, 7), sqlite3_column_int64(c->src, 8),
1217 53 : c->frag_checksum, c->frag_total, c->frag_part, c->frag_count);
1218 53 : cloudsync_memory_free(frag);
1219 53 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1220 53 : rc = cloudsync_payload_encode_final(payload, data);
1221 53 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1222 53 : c->payload = cloudsync_payload_blob(payload, &c->payload_size, &c->rows);
1223 53 : cloudsync_memory_free(payload);
1224 53 : c->dbv_min = sqlite3_column_int64(c->src, 5);
1225 53 : c->dbv_max = c->dbv_min;
1226 53 : c->chunk_index++;
1227 53 : c->frag_offset += frag_len;
1228 53 : c->frag_part++;
1229 53 : if (c->frag_part >= c->frag_count) {
1230 15 : c->frag_active = false;
1231 15 : rc = payload_chunks_step_source(c);
1232 15 : }
1233 53 : return rc;
1234 53 : }
1235 :
1236 94 : static int payload_chunks_build_next(cloudsync_payload_chunks_cursor *c) {
1237 94 : cloudsync_context *data = c->vtab->data;
1238 94 : int rc = SQLITE_OK;
1239 94 : if (c->payload) { cloudsync_memory_free(c->payload); c->payload = NULL; }
1240 94 : c->payload_size = c->rows = c->dbv_min = c->dbv_max = 0;
1241 94 : if (c->frag_active) return payload_chunks_emit_fragment(c);
1242 56 : if (!c->has_row) { c->eof = true; return SQLITE_OK; }
1243 :
1244 39 : int max_size = cloudsync_payload_max_chunk_size(data);
1245 39 : size_t payload_header_size = 0;
1246 39 : cloudsync_payload_context_size(&payload_header_size);
1247 39 : cloudsync_payload_context *payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
1248 39 : if (!payload) return SQLITE_NOMEM;
1249 2076 : while (c->has_row) {
1250 : sqlite3_value *rowv[9];
1251 20660 : for (int i = 0; i < 9; ++i) rowv[i] = sqlite3_column_value(c->src, i);
1252 2066 : size_t row_size = pk_encode_size((dbvalue_t **)rowv, 9, 0, -1);
1253 2066 : if (row_size == SIZE_MAX) { cloudsync_memory_free(payload); return SQLITE_NOMEM; }
1254 :
1255 2066 : if ((int64_t)row_size + (int64_t)payload_header_size + CLOUDSYNC_PAYLOAD_CHUNK_SAFETY_MARGIN > max_size) {
1256 21 : if (cloudsync_payload_context_nrows(payload) > 0) break;
1257 15 : cloudsync_memory_free(payload);
1258 15 : rc = payload_chunks_begin_fragment(c, 0);
1259 15 : if (rc != SQLITE_OK) return rc;
1260 15 : return payload_chunks_emit_fragment(c);
1261 : }
1262 :
1263 2045 : if (cloudsync_payload_context_nrows(payload) > 0 && cloudsync_payload_context_bused(payload) + row_size > (size_t)max_size) break;
1264 2037 : rc = cloudsync_payload_encode_step(payload, data, 9, (dbvalue_t **)rowv);
1265 2037 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1266 2037 : int64_t dbv = sqlite3_column_int64(c->src, 5);
1267 2037 : if (cloudsync_payload_context_nrows(payload) == 1) c->dbv_min = dbv;
1268 2037 : c->dbv_max = dbv;
1269 2037 : rc = payload_chunks_step_source(c);
1270 2037 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1271 : }
1272 :
1273 24 : if (cloudsync_payload_context_nrows(payload) == 0) { cloudsync_memory_free(payload); c->eof = true; return SQLITE_OK; }
1274 24 : rc = cloudsync_payload_encode_final(payload, data);
1275 24 : if (rc != SQLITE_OK) { cloudsync_memory_free(payload); return rc; }
1276 24 : c->payload = cloudsync_payload_blob(payload, &c->payload_size, &c->rows);
1277 24 : cloudsync_memory_free(payload);
1278 24 : c->chunk_index++;
1279 24 : return SQLITE_OK;
1280 94 : }
1281 :
1282 : // Record the resume point a stateless caller passes back to continue after the
1283 : // chunk just built. Reads the source statement, which is positioned at the next
1284 : // unconsumed row (or the same row when a value is still mid-fragment). Must be
1285 : // called only after build_next produced a chunk (i.e. !eof).
1286 77 : static void payload_chunks_set_next_cursor(cloudsync_payload_chunks_cursor *c) {
1287 77 : if (c->frag_active) {
1288 : // Mid-value: resume the same row at the next byte offset.
1289 38 : c->next_dbv = sqlite3_column_int64(c->src, 5);
1290 38 : c->next_seq = sqlite3_column_int64(c->src, 8);
1291 38 : c->next_frag_offset = c->frag_offset;
1292 38 : c->is_final = false;
1293 77 : } else if (c->has_row) {
1294 : // Row boundary: the next chunk starts at the current (unconsumed) row.
1295 22 : c->next_dbv = sqlite3_column_int64(c->src, 5);
1296 22 : c->next_seq = sqlite3_column_int64(c->src, 8);
1297 22 : c->next_frag_offset = 0;
1298 22 : c->is_final = false;
1299 22 : } else {
1300 : // Stream exhausted: this was the last chunk of the window.
1301 17 : c->next_dbv = c->watermark;
1302 17 : c->next_seq = 0;
1303 17 : c->next_frag_offset = 0;
1304 17 : c->is_final = true;
1305 : }
1306 77 : }
1307 :
1308 94 : static int payload_chunks_advance(cloudsync_payload_chunks_cursor *c) {
1309 94 : int rc = payload_chunks_build_next(c);
1310 94 : if (rc == SQLITE_OK && !c->eof) payload_chunks_set_next_cursor(c);
1311 94 : return rc;
1312 : }
1313 :
1314 29 : static int payload_chunks_filter(sqlite3_vtab_cursor *cursor, int idxnum, const char *idxstr, int argc, sqlite3_value **argv) {
1315 29 : UNUSED_PARAMETER(idxstr); UNUSED_PARAMETER(argc);
1316 29 : cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
1317 29 : cloudsync_context *data = c->vtab->data;
1318 29 : if (!cloudsync_context_is_initialized(data)) {
1319 1 : c->vtab->base.zErrMsg = sqlite3_mprintf(
1320 : "cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') "
1321 : "to enable sync on a table before querying cloudsync_payload_chunks.");
1322 1 : return SQLITE_ERROR;
1323 : }
1324 28 : if (c->src) { sqlite3_finalize(c->src); c->src = NULL; }
1325 28 : if (c->payload) { cloudsync_memory_free(c->payload); c->payload = NULL; }
1326 : // Contract: all per-scan state that can be bulk-reset here must live at or
1327 : // after eof. Fields before eof are cursor lifetime state preserved across
1328 : // xFilter calls.
1329 28 : memset(&c->eof, 0, sizeof(*c) - offsetof(cloudsync_payload_chunks_cursor, eof));
1330 :
1331 28 : int argi = 0;
1332 28 : int64_t since = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_SEND_DBVERSION);
1333 28 : const void *site_id = NULL;
1334 28 : int site_id_len = 0;
1335 28 : bool site_id_given = false;
1336 28 : int64_t until = 0;
1337 28 : bool exclude = false;
1338 : // Positional resume cursor (cols 15..17): when resume_db_version is bound the
1339 : // scan starts at (resume_db_version, resume_seq) inclusive and the first chunk
1340 : // resumes a mid-value fragment at resume_frag_offset, instead of replaying the
1341 : // whole window from `since`. Lets a stateless /check page the stream with an
1342 : // O(1) seek per call and no spool table.
1343 28 : bool positional = false;
1344 28 : int64_t resume_dbv = 0, resume_seq = 0, resume_frag = 0;
1345 28 : if (idxnum & 1) since = sqlite3_value_int64(argv[argi++]);
1346 28 : if (idxnum & 2) {
1347 2 : if (sqlite3_value_type(argv[argi]) != SQLITE_NULL) {
1348 2 : site_id = sqlite3_value_blob(argv[argi]);
1349 2 : site_id_len = sqlite3_value_bytes(argv[argi]);
1350 2 : site_id_given = true;
1351 2 : }
1352 2 : argi++;
1353 2 : }
1354 28 : if (idxnum & 4) until = sqlite3_value_int64(argv[argi++]);
1355 28 : if (idxnum & 8) exclude = (sqlite3_value_int(argv[argi++]) != 0);
1356 28 : if (idxnum & 16) { resume_dbv = sqlite3_value_int64(argv[argi++]); positional = true; }
1357 28 : if (idxnum & 32) resume_seq = sqlite3_value_int64(argv[argi++]);
1358 28 : if (idxnum & 64) resume_frag = sqlite3_value_int64(argv[argi++]);
1359 :
1360 : // Resolve the site filter:
1361 : // exclude=true -> all sites except filter_site_id (CHECK path); site required
1362 : // filter given -> only that site
1363 : // default -> local site (send path, unchanged)
1364 : const char *site_op;
1365 28 : if (exclude) {
1366 2 : if (!site_id_given) {
1367 1 : c->vtab->base.zErrMsg = sqlite3_mprintf(
1368 : "cloudsync_payload_chunks: exclude_filter_site_id requires a non-NULL site_id");
1369 1 : return SQLITE_ERROR;
1370 : }
1371 1 : site_op = "<>";
1372 1 : } else {
1373 26 : site_op = "=";
1374 26 : if (!site_id_given) { site_id = cloudsync_siteid(data); site_id_len = UUID_LEN; }
1375 : }
1376 :
1377 27 : if (until == 0) {
1378 17 : char *mxsql = sqlite3_mprintf(
1379 17 : "SELECT COALESCE(MAX(db_version),0) FROM cloudsync_changes WHERE site_id%s?", site_op);
1380 17 : if (!mxsql) return SQLITE_NOMEM;
1381 17 : sqlite3_stmt *mx = NULL;
1382 17 : int rc = sqlite3_prepare_v2(c->vtab->db, mxsql, -1, &mx, NULL);
1383 17 : sqlite3_free(mxsql);
1384 17 : if (rc != SQLITE_OK) return rc;
1385 17 : sqlite3_bind_blob(mx, 1, site_id, site_id_len, SQLITE_TRANSIENT);
1386 : // MAX() yields exactly one row, so anything else is an error: propagate
1387 : // it instead of silently scanning an empty window with until=0
1388 17 : rc = sqlite3_step(mx);
1389 17 : if (rc == SQLITE_ROW) until = sqlite3_column_int64(mx, 0);
1390 17 : sqlite3_finalize(mx);
1391 17 : if (rc != SQLITE_ROW) {
1392 0 : if (c->vtab->base.zErrMsg) sqlite3_free(c->vtab->base.zErrMsg);
1393 0 : c->vtab->base.zErrMsg = sqlite3_mprintf("%s", sqlite3_errmsg(c->vtab->db));
1394 0 : return rc == SQLITE_DONE ? SQLITE_ERROR : rc;
1395 : }
1396 17 : }
1397 27 : c->watermark = until;
1398 :
1399 : // Window upper bound is always `until`. The lower bound is either the legacy
1400 : // exclusive `since` (db_version > since) or the inclusive positional cursor
1401 : // (db_version, seq) >= (resume_dbv, resume_seq).
1402 : char *sql;
1403 27 : if (positional) {
1404 : // The redundant db_version>=? is what makes the resume a seek, and it is
1405 : // load-bearing: SQLite derives a range from a disjunction only when both arms
1406 : // compare against the same value, and these two arms carry distinct
1407 : // parameters. Without it cloudsync_changes' xBestIndex is offered no lower
1408 : // bound at all and every call replays the window from the start.
1409 10 : sql = sqlite3_mprintf(
1410 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1411 : "FROM cloudsync_changes WHERE db_version<=? AND site_id%s? AND db_version>=? AND "
1412 : "(db_version>? OR (db_version=? AND seq>=?)) ORDER BY db_version, seq ASC",
1413 10 : site_op);
1414 10 : } else {
1415 17 : sql = sqlite3_mprintf(
1416 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1417 : "FROM cloudsync_changes WHERE db_version>? AND site_id%s? AND db_version<=? ORDER BY db_version, seq ASC",
1418 17 : site_op);
1419 : }
1420 27 : if (!sql) return SQLITE_NOMEM;
1421 27 : int rc = sqlite3_prepare_v2(c->vtab->db, sql, -1, &c->src, NULL);
1422 27 : sqlite3_free(sql);
1423 27 : if (rc != SQLITE_OK) return rc;
1424 27 : if (positional) {
1425 10 : sqlite3_bind_int64(c->src, 1, until);
1426 10 : sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT);
1427 10 : sqlite3_bind_int64(c->src, 3, resume_dbv);
1428 10 : sqlite3_bind_int64(c->src, 4, resume_dbv);
1429 10 : sqlite3_bind_int64(c->src, 5, resume_dbv);
1430 10 : sqlite3_bind_int64(c->src, 6, resume_seq);
1431 10 : } else {
1432 17 : sqlite3_bind_int64(c->src, 1, since);
1433 17 : sqlite3_bind_blob(c->src, 2, site_id, site_id_len, SQLITE_TRANSIENT);
1434 17 : sqlite3_bind_int64(c->src, 3, until);
1435 : }
1436 27 : rc = payload_chunks_step_source(c);
1437 27 : if (rc != SQLITE_OK) return rc;
1438 : // Resuming inside a value that was fragmented across chunks: the first row is
1439 : // that value; re-establish the fragment plan and skip to resume_frag.
1440 27 : if (positional && resume_frag > 0 && c->has_row) {
1441 6 : rc = payload_chunks_begin_fragment(c, resume_frag);
1442 6 : if (rc != SQLITE_OK) return rc;
1443 6 : }
1444 27 : return payload_chunks_advance(c);
1445 29 : }
1446 :
1447 67 : static int payload_chunks_next(sqlite3_vtab_cursor *cursor) {
1448 67 : return payload_chunks_advance((cloudsync_payload_chunks_cursor *)cursor);
1449 : }
1450 :
1451 94 : static int payload_chunks_eof(sqlite3_vtab_cursor *cursor) {
1452 94 : return ((cloudsync_payload_chunks_cursor *)cursor)->eof;
1453 : }
1454 :
1455 247 : static int payload_chunks_column(sqlite3_vtab_cursor *cursor, sqlite3_context *ctx, int col) {
1456 247 : cloudsync_payload_chunks_cursor *c = (cloudsync_payload_chunks_cursor *)cursor;
1457 247 : switch (col) {
1458 91 : case 0: sqlite3_result_blob64(ctx, c->payload, (sqlite3_uint64)c->payload_size, SQLITE_TRANSIENT); break;
1459 63 : case 1: sqlite3_result_int(ctx, c->chunk_index - 1); break;
1460 17 : case 2: sqlite3_result_int64(ctx, c->payload_size); break;
1461 17 : case 3: sqlite3_result_int64(ctx, c->rows); break;
1462 3 : case 4: sqlite3_result_int64(ctx, c->dbv_min); break;
1463 3 : case 5: sqlite3_result_int64(ctx, c->dbv_max); break;
1464 13 : case 6: sqlite3_result_int64(ctx, c->watermark); break;
1465 10 : case 11: sqlite3_result_int64(ctx, c->next_dbv); break;
1466 10 : case 12: sqlite3_result_int64(ctx, c->next_seq); break;
1467 10 : case 13: sqlite3_result_int64(ctx, c->next_frag_offset); break;
1468 10 : case 14: sqlite3_result_int(ctx, c->is_final ? 1 : 0); break;
1469 0 : default: sqlite3_result_null(ctx); break;
1470 : }
1471 247 : return SQLITE_OK;
1472 : }
1473 :
1474 0 : static int payload_chunks_rowid(sqlite3_vtab_cursor *cursor, sqlite3_int64 *rowid) {
1475 0 : *rowid = ((cloudsync_payload_chunks_cursor *)cursor)->chunk_index;
1476 0 : return SQLITE_OK;
1477 : }
1478 :
1479 : static sqlite3_module cloudsync_payload_chunks_module = {
1480 : /* iVersion */ 0,
1481 : /* xCreate */ NULL,
1482 : /* xConnect */ payload_chunks_connect,
1483 : /* xBestIndex */ payload_chunks_best_index,
1484 : /* xDisconnect */ payload_chunks_disconnect,
1485 : /* xDestroy */ NULL,
1486 : /* xOpen */ payload_chunks_open,
1487 : /* xClose */ payload_chunks_close,
1488 : /* xFilter */ payload_chunks_filter,
1489 : /* xNext */ payload_chunks_next,
1490 : /* xEof */ payload_chunks_eof,
1491 : /* xColumn */ payload_chunks_column,
1492 : /* xRowid */ payload_chunks_rowid,
1493 : /* xUpdate */ NULL,
1494 : /* xBegin */ NULL,
1495 : /* xSync */ NULL,
1496 : /* xCommit */ NULL,
1497 : /* xRollback */ NULL,
1498 : /* xFindMethod */ NULL,
1499 : /* xRename */ NULL,
1500 : /* xSavepoint */ NULL,
1501 : /* xRelease */ NULL,
1502 : /* xRollbackTo */ NULL,
1503 : /* xShadowName */ NULL,
1504 : /* xIntegrity */ NULL
1505 : };
1506 :
1507 1060 : static int payload_estimated_size_add(sqlite3_int64 *acc, sqlite3_int64 value) {
1508 1060 : if (value < 0 || *acc > INT64_MAX - value) return SQLITE_TOOBIG;
1509 1060 : *acc += value;
1510 1060 : return SQLITE_OK;
1511 1060 : }
1512 :
1513 6 : static int payload_blob_checked_estimate(sqlite3 *db, const void *site_id, int site_id_len,
1514 : sqlite3_int64 since, sqlite3_int64 since_seq,
1515 : bool exclude, sqlite3_int64 *estimated,
1516 : sqlite3_int64 *watermark) {
1517 6 : sqlite3_stmt *stmt = NULL;
1518 6 : sqlite3_stmt *mx = NULL;
1519 6 : int rc = SQLITE_OK;
1520 6 : sqlite3_int64 until = 0;
1521 6 : size_t header_size = 0;
1522 6 : bool has_rows = false;
1523 6 : const char *site_op = exclude ? "<>" : "=";
1524 6 : *estimated = 0;
1525 :
1526 6 : char *mxsql = sqlite3_mprintf(
1527 6 : "SELECT COALESCE(MAX(db_version),0) FROM cloudsync_changes WHERE site_id%s?", site_op);
1528 6 : if (!mxsql) return SQLITE_NOMEM;
1529 6 : rc = sqlite3_prepare_v2(db, mxsql, -1, &mx, NULL);
1530 6 : sqlite3_free(mxsql);
1531 6 : if (rc != SQLITE_OK) goto error;
1532 6 : sqlite3_bind_blob(mx, 1, site_id, site_id_len, SQLITE_TRANSIENT);
1533 6 : rc = sqlite3_step(mx);
1534 6 : if (rc == SQLITE_ROW) until = sqlite3_column_int64(mx, 0);
1535 0 : else if (rc != SQLITE_DONE) goto error;
1536 6 : sqlite3_finalize(mx);
1537 6 : mx = NULL;
1538 6 : if (watermark) *watermark = until;
1539 :
1540 6 : if (until < since) return SQLITE_OK;
1541 :
1542 4 : char *sql = sqlite3_mprintf(
1543 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1544 : "FROM cloudsync_changes WHERE (db_version>? OR (db_version=? AND seq>?)) "
1545 : "AND site_id%s? AND db_version<=? ORDER BY db_version, seq ASC",
1546 4 : site_op);
1547 4 : if (!sql) return SQLITE_NOMEM;
1548 4 : rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
1549 4 : sqlite3_free(sql);
1550 4 : if (rc != SQLITE_OK) goto error;
1551 4 : sqlite3_bind_int64(stmt, 1, since);
1552 4 : sqlite3_bind_int64(stmt, 2, since);
1553 4 : sqlite3_bind_int64(stmt, 3, since_seq);
1554 4 : sqlite3_bind_blob(stmt, 4, site_id, site_id_len, SQLITE_TRANSIENT);
1555 4 : sqlite3_bind_int64(stmt, 5, until);
1556 :
1557 4 : cloudsync_payload_context_size(&header_size);
1558 1060 : while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
1559 : sqlite3_value *rowv[9];
1560 10560 : for (int i = 0; i < 9; ++i) rowv[i] = sqlite3_column_value(stmt, i);
1561 1056 : size_t row_size = pk_encode_size((dbvalue_t **)rowv, 9, 0, -1);
1562 1056 : if (row_size == SIZE_MAX || row_size > (size_t)INT64_MAX) {
1563 0 : rc = SQLITE_TOOBIG;
1564 0 : goto error;
1565 : }
1566 1056 : if (!has_rows) {
1567 4 : if (header_size > (size_t)INT64_MAX) {
1568 0 : rc = SQLITE_TOOBIG;
1569 0 : goto error;
1570 : }
1571 4 : rc = payload_estimated_size_add(estimated, (sqlite3_int64)header_size);
1572 4 : if (rc != SQLITE_OK) goto error;
1573 4 : has_rows = true;
1574 4 : }
1575 1056 : rc = payload_estimated_size_add(estimated, (sqlite3_int64)row_size);
1576 1056 : if (rc != SQLITE_OK) goto error;
1577 : }
1578 4 : if (rc != SQLITE_DONE) goto error;
1579 4 : sqlite3_finalize(stmt);
1580 4 : return SQLITE_OK;
1581 :
1582 : error:
1583 0 : if (stmt) sqlite3_finalize(stmt);
1584 0 : if (mx) sqlite3_finalize(mx);
1585 0 : return rc;
1586 6 : }
1587 :
1588 7 : void dbsync_payload_blob_checked(sqlite3_context *context, int argc, sqlite3_value **argv) {
1589 : DEBUG_FUNCTION("cloudsync_payload_blob_checked");
1590 7 : UNUSED_PARAMETER(argc);
1591 :
1592 7 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1593 7 : sqlite3 *db = sqlite3_context_db_handle(context);
1594 7 : sqlite3_stmt *stmt = NULL;
1595 7 : cloudsync_payload_context *payload = NULL;
1596 7 : int rc = SQLITE_OK;
1597 7 : sqlite3_int64 since = 0;
1598 7 : sqlite3_int64 since_seq = 0;
1599 7 : sqlite3_int64 max_estimated_size = 0;
1600 7 : sqlite3_int64 estimated = 0;
1601 7 : sqlite3_int64 watermark = 0;
1602 7 : const void *site_id = NULL;
1603 7 : int site_id_len = 0;
1604 7 : bool exclude = false;
1605 :
1606 7 : if (sqlite3_value_type(argv[0]) != SQLITE_NULL) since = sqlite3_value_int64(argv[0]);
1607 7 : if (sqlite3_value_type(argv[1]) != SQLITE_NULL) since_seq = sqlite3_value_int64(argv[1]);
1608 7 : if (sqlite3_value_type(argv[2]) != SQLITE_NULL) {
1609 3 : site_id = sqlite3_value_blob(argv[2]);
1610 3 : site_id_len = sqlite3_value_bytes(argv[2]);
1611 3 : }
1612 7 : exclude = sqlite3_value_type(argv[3]) != SQLITE_NULL && sqlite3_value_int(argv[3]) != 0;
1613 7 : if (sqlite3_value_type(argv[4]) == SQLITE_NULL || sqlite3_value_int64(argv[4]) <= 0) {
1614 0 : sqlite3_result_error(context, "cloudsync_payload_blob_checked: max_estimated_payload_size must be positive", -1);
1615 0 : return;
1616 : }
1617 7 : max_estimated_size = sqlite3_value_int64(argv[4]);
1618 :
1619 7 : if (exclude && !site_id) {
1620 1 : sqlite3_result_error(context,
1621 : "cloudsync_payload_blob_checked: exclude_filter_site_id requires a non-NULL site_id", -1);
1622 1 : return;
1623 : }
1624 6 : if (!exclude && !site_id) {
1625 3 : site_id = cloudsync_siteid(data);
1626 3 : site_id_len = UUID_LEN;
1627 3 : }
1628 :
1629 6 : rc = payload_blob_checked_estimate(db, site_id, site_id_len, since, since_seq, exclude, &estimated, &watermark);
1630 6 : if (rc != SQLITE_OK) goto error;
1631 6 : if (estimated == 0) {
1632 2 : sqlite3_result_null(context);
1633 2 : return;
1634 : }
1635 4 : if (estimated > max_estimated_size) {
1636 2 : dbsync_set_error(context,
1637 : CLOUDSYNC_ERRCODE_PAYLOAD_TOO_LARGE "cloudsync_payload_blob_checked: estimated payload size %" PRId64 " exceeds max_estimated_payload_size %" PRId64,
1638 1 : (int64_t)estimated, (int64_t)max_estimated_size);
1639 1 : return;
1640 : }
1641 :
1642 3 : const char *site_op = exclude ? "<>" : "=";
1643 3 : char *sql = sqlite3_mprintf(
1644 : "SELECT tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq "
1645 : "FROM cloudsync_changes WHERE (db_version>? OR (db_version=? AND seq>?)) "
1646 : "AND site_id%s? AND db_version<=? ORDER BY db_version, seq ASC",
1647 3 : site_op);
1648 3 : if (!sql) { rc = SQLITE_NOMEM; goto error; }
1649 3 : rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
1650 3 : sqlite3_free(sql);
1651 3 : if (rc != SQLITE_OK) goto error;
1652 3 : sqlite3_bind_int64(stmt, 1, since);
1653 3 : sqlite3_bind_int64(stmt, 2, since);
1654 3 : sqlite3_bind_int64(stmt, 3, since_seq);
1655 3 : sqlite3_bind_blob(stmt, 4, site_id, site_id_len, SQLITE_TRANSIENT);
1656 3 : sqlite3_bind_int64(stmt, 5, watermark);
1657 :
1658 3 : payload = cloudsync_memory_zeroalloc((uint64_t)cloudsync_payload_context_size(NULL));
1659 3 : if (!payload) { rc = SQLITE_NOMEM; goto error; }
1660 :
1661 533 : while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
1662 : sqlite3_value *rowv[9];
1663 5300 : for (int i = 0; i < 9; ++i) rowv[i] = sqlite3_column_value(stmt, i);
1664 530 : rc = cloudsync_payload_encode_step(payload, data, 9, (dbvalue_t **)rowv);
1665 530 : if (rc != SQLITE_OK) goto error;
1666 : }
1667 3 : if (rc != SQLITE_DONE) goto error;
1668 3 : sqlite3_finalize(stmt);
1669 3 : stmt = NULL;
1670 :
1671 3 : rc = cloudsync_payload_encode_final(payload, data);
1672 3 : if (rc != SQLITE_OK) goto error;
1673 3 : int64_t blob_size = 0;
1674 3 : char *blob = cloudsync_payload_blob(payload, &blob_size, NULL);
1675 3 : if (!blob) {
1676 0 : sqlite3_result_null(context);
1677 0 : cloudsync_payload_context_free(payload);
1678 0 : } else {
1679 3 : sqlite3_result_blob64(context, blob, (sqlite3_uint64)blob_size, cloudsync_memory_free);
1680 3 : cloudsync_memory_free(payload);
1681 : }
1682 3 : return;
1683 :
1684 : error:
1685 0 : if (stmt) sqlite3_finalize(stmt);
1686 0 : if (payload) cloudsync_payload_context_free(payload);
1687 0 : if (rc == SQLITE_NOMEM) sqlite3_result_error_nomem(context);
1688 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);
1689 0 : else sqlite3_result_error(context, sqlite3_errmsg(db), -1);
1690 7 : }
1691 :
1692 : #ifdef CLOUDSYNC_DESKTOP_OS
1693 0 : void dbsync_payload_save (sqlite3_context *context, int argc, sqlite3_value **argv) {
1694 : DEBUG_FUNCTION("dbsync_payload_save");
1695 :
1696 : // sanity check argument
1697 0 : if (database_value_type(argv[0]) != SQLITE_TEXT) {
1698 0 : sqlite3_result_error(context, "Unable to retrieve file path.", -1);
1699 0 : return;
1700 : }
1701 :
1702 : // retrieve full path to file
1703 0 : const char *payload_path = (const char *)database_value_text(argv[0]);
1704 :
1705 : // retrieve global context
1706 0 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1707 :
1708 0 : int blob_size = 0;
1709 0 : int rc = cloudsync_payload_save(data, payload_path, &blob_size);
1710 0 : if (rc == SQLITE_OK) {
1711 : // if OK then returns blob size
1712 0 : sqlite3_result_int64(context, (sqlite3_int64)blob_size);
1713 0 : return;
1714 : }
1715 :
1716 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
1717 0 : sqlite3_result_error_code(context, rc);
1718 0 : }
1719 :
1720 0 : void dbsync_payload_load (sqlite3_context *context, int argc, sqlite3_value **argv) {
1721 : DEBUG_FUNCTION("dbsync_payload_load");
1722 :
1723 : // sanity check argument
1724 0 : if (database_value_type(argv[0]) != SQLITE_TEXT) {
1725 0 : sqlite3_result_error(context, "Unable to retrieve file path.", -1);
1726 0 : return;
1727 : }
1728 :
1729 : // retrieve full path to file
1730 0 : const char *path = (const char *)database_value_text(argv[0]);
1731 :
1732 0 : int64_t payload_size = 0;
1733 0 : char *payload = cloudsync_file_read(path, &payload_size);
1734 0 : if (!payload) {
1735 0 : if (payload_size < 0) {
1736 0 : sqlite3_result_error(context, "Unable to read payload from file path.", -1);
1737 0 : sqlite3_result_error_code(context, SQLITE_IOERR);
1738 0 : return;
1739 : }
1740 : // no rows affected but no error either
1741 0 : sqlite3_result_int(context, 0);
1742 0 : return;
1743 : }
1744 :
1745 0 : int nrows = 0;
1746 0 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1747 : // File-based load applies a complete monolithic payload: legacy last-applied
1748 : // checkpoint (ends on a db_version boundary, so it is safe).
1749 0 : int rc = cloudsync_payload_apply (data, payload, (int)payload_size, &nrows, CLOUDSYNC_CHECKPOINT_LAST_APPLIED, 0);
1750 0 : if (payload) cloudsync_memory_free(payload);
1751 :
1752 0 : if (rc != SQLITE_OK) {
1753 0 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
1754 0 : sqlite3_result_error_code(context, rc);
1755 0 : return;
1756 : }
1757 :
1758 : // returns number of applied rows
1759 0 : sqlite3_result_int(context, nrows);
1760 0 : }
1761 : #endif
1762 :
1763 : // MARK: - Register -
1764 :
1765 31878 : 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 *)) {
1766 :
1767 31878 : int rc = sqlite3_create_function_v2(db, name, nargs, flags, ctx, xfunc, xstep, xfinal, ctx_free);
1768 :
1769 31878 : if (rc != SQLITE_OK) {
1770 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("Error creating function %s: %s", name, sqlite3_errmsg(db));
1771 0 : return rc;
1772 : }
1773 31878 : return SQLITE_OK;
1774 31878 : }
1775 :
1776 26565 : 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 *)) {
1777 26565 : const int FLAGS_VOLATILE = SQLITE_UTF8;
1778 : DEBUG_DBFUNCTION("dbsync_register %s", name);
1779 26565 : return dbsync_register_with_flags(db, name, xfunc, xstep, xfinal, nargs, FLAGS_VOLATILE, pzErrMsg, ctx, ctx_free);
1780 : }
1781 :
1782 25806 : 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 *)) {
1783 : DEBUG_DBFUNCTION("dbsync_register_function %s", name);
1784 25806 : return dbsync_register(db, name, xfunc, NULL, NULL, nargs, pzErrMsg, ctx, ctx_free);
1785 : }
1786 :
1787 2277 : 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 *)) {
1788 2277 : const int FLAGS_PURE = SQLITE_UTF8 | SQLITE_INNOCUOUS | SQLITE_DETERMINISTIC;
1789 : DEBUG_DBFUNCTION("dbsync_register_pure_function %s", name);
1790 2277 : return dbsync_register_with_flags(db, name, xfunc, NULL, NULL, nargs, FLAGS_PURE, pzErrMsg, ctx, ctx_free);
1791 : }
1792 :
1793 2277 : 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 *)) {
1794 2277 : const int FLAGS_TRIGGER = SQLITE_UTF8 | SQLITE_INNOCUOUS;
1795 : DEBUG_DBFUNCTION("dbsync_register_trigger_function %s", name);
1796 2277 : return dbsync_register_with_flags(db, name, xfunc, NULL, NULL, nargs, FLAGS_TRIGGER, pzErrMsg, ctx, ctx_free);
1797 : }
1798 :
1799 759 : 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 *)) {
1800 : DEBUG_DBFUNCTION("dbsync_register_aggregate %s", name);
1801 759 : return dbsync_register(db, name, NULL, xstep, xfinal, nargs, pzErrMsg, ctx, ctx_free);
1802 : }
1803 :
1804 759 : 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 *)) {
1805 759 : const int FLAGS_TRIGGER = SQLITE_UTF8 | SQLITE_INNOCUOUS;
1806 : DEBUG_DBFUNCTION("dbsync_register_trigger_aggregate %s", name);
1807 759 : return dbsync_register_with_flags(db, name, NULL, xstep, xfinal, nargs, FLAGS_TRIGGER, pzErrMsg, ctx, ctx_free);
1808 : }
1809 :
1810 : // MARK: - Block-level LWW -
1811 :
1812 603 : void dbsync_text_materialize (sqlite3_context *context, int argc, sqlite3_value **argv) {
1813 : DEBUG_FUNCTION("cloudsync_text_materialize");
1814 :
1815 : // argv[0] -> table name
1816 : // argv[1] -> column name
1817 : // argv[2..N] -> primary key values
1818 :
1819 603 : if (argc < 3) {
1820 0 : sqlite3_result_error(context, "cloudsync_text_materialize requires at least 3 arguments: table, column, pk...", -1);
1821 0 : return;
1822 : }
1823 :
1824 603 : const char *table_name = (const char *)database_value_text(argv[0]);
1825 603 : const char *col_name = (const char *)database_value_text(argv[1]);
1826 603 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1827 :
1828 603 : cloudsync_table_context *table = table_lookup(data, table_name);
1829 603 : if (!table) {
1830 0 : dbsync_set_error(context, "Unable to retrieve table name %s in cloudsync_text_materialize.", table_name);
1831 0 : return;
1832 : }
1833 :
1834 603 : int col_idx = table_col_index(table, col_name);
1835 603 : if (col_idx < 0 || table_col_algo(table, col_idx) != col_algo_block) {
1836 0 : dbsync_set_error(context, "Column %s in table %s is not configured as block-level.", col_name, table_name);
1837 0 : return;
1838 : }
1839 :
1840 : // Encode primary keys
1841 603 : int npks = table_count_pks(table);
1842 603 : if (argc - 2 != npks) {
1843 0 : sqlite3_result_error(context, "Wrong number of primary key values for cloudsync_text_materialize.", -1);
1844 0 : return;
1845 : }
1846 :
1847 : char buffer[1024];
1848 603 : size_t pklen = sizeof(buffer);
1849 603 : char *pk = pk_encode_prikey((dbvalue_t **)&argv[2], npks, buffer, &pklen);
1850 603 : if (!pk || pk == PRIKEY_NULL_CONSTRAINT_ERROR) {
1851 0 : sqlite3_result_error(context, "Failed to encode primary key(s).", -1);
1852 0 : return;
1853 : }
1854 :
1855 : // Materialize the column
1856 603 : int rc = block_materialize_column(data, table, pk, (int)pklen, col_name);
1857 603 : if (rc != DBRES_OK) {
1858 309 : sqlite3_result_error(context, cloudsync_errmsg(data), -1);
1859 309 : sqlite3_result_error_code(context, rc);
1860 309 : } else {
1861 294 : sqlite3_result_int(context, 1);
1862 : }
1863 :
1864 603 : if (pk != buffer) cloudsync_memory_free(pk);
1865 603 : }
1866 :
1867 : // MARK: - Row Filter -
1868 :
1869 21 : void dbsync_set_filter (sqlite3_context *context, int argc, sqlite3_value **argv) {
1870 : DEBUG_FUNCTION("cloudsync_set_filter");
1871 :
1872 21 : const char *tbl = (const char *)database_value_text(argv[0]);
1873 21 : const char *filter_expr = (const char *)database_value_text(argv[1]);
1874 21 : if (!tbl || !filter_expr) {
1875 0 : dbsync_set_error(context, "cloudsync_set_filter: table and filter expression required");
1876 0 : return;
1877 : }
1878 :
1879 21 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1880 :
1881 : // Guard against calling set_filter before the target table has been set
1882 : // up for sync: without this, we'd hit "no such table:
1883 : // cloudsync_table_settings" or "no such table: main.<tbl>" deep inside
1884 : // the trigger recreation path, which is not actionable.
1885 21 : if (!cloudsync_context_is_initialized(data) || !table_lookup(data, tbl)) {
1886 2 : dbsync_set_error(context,
1887 : "cloudsync_set_filter: table '%s' is not configured for sync. "
1888 1 : "Call SELECT cloudsync_init('%s') first.", tbl, tbl);
1889 1 : return;
1890 : }
1891 :
1892 : // Store filter in table settings
1893 20 : dbutils_table_settings_set_key_value(data, tbl, "*", "filter", filter_expr);
1894 :
1895 : // Read current algo
1896 20 : table_algo algo = dbutils_table_settings_get_algo(data, tbl);
1897 20 : if (algo == table_algo_none) algo = table_algo_crdt_cls;
1898 :
1899 : // Drop and recreate triggers with the filter
1900 20 : database_delete_triggers(data, tbl);
1901 20 : int rc = database_create_triggers(data, tbl, algo, filter_expr);
1902 20 : if (rc != DBRES_OK) {
1903 0 : dbsync_set_error(context, "cloudsync_set_filter: error recreating triggers");
1904 0 : sqlite3_result_error_code(context, rc);
1905 0 : return;
1906 : }
1907 :
1908 : // Clean and refill metatable with the new filter
1909 20 : rc = cloudsync_reset_metatable(data, tbl);
1910 20 : if (rc != DBRES_OK) {
1911 0 : dbsync_set_error(context, "cloudsync_set_filter: error resetting metatable");
1912 0 : sqlite3_result_error_code(context, rc);
1913 0 : return;
1914 : }
1915 :
1916 20 : sqlite3_result_int(context, 1);
1917 21 : }
1918 :
1919 5 : void dbsync_clear_filter (sqlite3_context *context, int argc, sqlite3_value **argv) {
1920 : DEBUG_FUNCTION("cloudsync_clear_filter");
1921 :
1922 5 : const char *tbl = (const char *)database_value_text(argv[0]);
1923 5 : if (!tbl) {
1924 0 : dbsync_set_error(context, "cloudsync_clear_filter: table name required");
1925 0 : return;
1926 : }
1927 :
1928 5 : cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
1929 :
1930 : // Guard against calling clear_filter before the target table has been set
1931 : // up for sync — see dbsync_set_filter for the same rationale.
1932 5 : if (!cloudsync_context_is_initialized(data) || !table_lookup(data, tbl)) {
1933 2 : dbsync_set_error(context,
1934 : "cloudsync_clear_filter: table '%s' is not configured for sync. "
1935 1 : "Call SELECT cloudsync_init('%s') first.", tbl, tbl);
1936 1 : return;
1937 : }
1938 :
1939 : // Remove filter from table settings (set to NULL/empty)
1940 4 : dbutils_table_settings_set_key_value(data, tbl, "*", "filter", NULL);
1941 :
1942 : // Read current algo
1943 4 : table_algo algo = dbutils_table_settings_get_algo(data, tbl);
1944 4 : if (algo == table_algo_none) algo = table_algo_crdt_cls;
1945 :
1946 : // Drop and recreate triggers without filter
1947 4 : database_delete_triggers(data, tbl);
1948 4 : int rc = database_create_triggers(data, tbl, algo, NULL);
1949 4 : if (rc != DBRES_OK) {
1950 0 : dbsync_set_error(context, "cloudsync_clear_filter: error recreating triggers");
1951 0 : sqlite3_result_error_code(context, rc);
1952 0 : return;
1953 : }
1954 :
1955 : // Clean and refill metatable without filter (all rows)
1956 4 : rc = cloudsync_reset_metatable(data, tbl);
1957 4 : if (rc != DBRES_OK) {
1958 0 : dbsync_set_error(context, "cloudsync_clear_filter: error resetting metatable");
1959 0 : sqlite3_result_error_code(context, rc);
1960 0 : return;
1961 : }
1962 :
1963 4 : sqlite3_result_int(context, 1);
1964 5 : }
1965 :
1966 761 : int dbsync_register_functions (sqlite3 *db, char **pzErrMsg) {
1967 761 : int rc = SQLITE_OK;
1968 :
1969 : // there's no built-in way to verify if sqlite3_cloudsync_init has already been called
1970 : // for this specific database connection, we use a workaround: we attempt to retrieve the
1971 : // cloudsync_version and check for an error, an error indicates that initialization has not been performed
1972 761 : if (sqlite3_exec(db, "SELECT cloudsync_version();", NULL, NULL, NULL) == SQLITE_OK) return SQLITE_OK;
1973 :
1974 : // init memory debugger (NOOP in production)
1975 : cloudsync_memory_init(1);
1976 :
1977 : // set fractional-indexing allocator to use cloudsync memory
1978 759 : block_init_allocator();
1979 :
1980 : // init context
1981 759 : void *ctx = cloudsync_context_create(db);
1982 759 : if (!ctx) {
1983 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("Not enought memory to create a database context");
1984 0 : return SQLITE_NOMEM;
1985 : }
1986 :
1987 : // register functions
1988 :
1989 : // PUBLIC functions
1990 759 : rc = dbsync_register_pure_function(db, "cloudsync_version", dbsync_version, 0, pzErrMsg, ctx, cloudsync_context_free);
1991 759 : if (rc != SQLITE_OK) return rc;
1992 :
1993 759 : rc = dbsync_register_function(db, "cloudsync_init", dbsync_init1, 1, pzErrMsg, ctx, NULL);
1994 759 : if (rc != SQLITE_OK) return rc;
1995 :
1996 759 : rc = dbsync_register_function(db, "cloudsync_init", dbsync_init2, 2, pzErrMsg, ctx, NULL);
1997 759 : if (rc != SQLITE_OK) return rc;
1998 :
1999 759 : rc = dbsync_register_function(db, "cloudsync_init", dbsync_init3, 3, pzErrMsg, ctx, NULL);
2000 759 : if (rc != SQLITE_OK) return rc;
2001 :
2002 759 : rc = dbsync_register_function(db, "cloudsync_enable", dbsync_enable, 1, pzErrMsg, ctx, NULL);
2003 759 : if (rc != SQLITE_OK) return rc;
2004 :
2005 759 : rc = dbsync_register_function(db, "cloudsync_disable", dbsync_disable, 1, pzErrMsg, ctx, NULL);
2006 759 : if (rc != SQLITE_OK) return rc;
2007 :
2008 759 : rc = dbsync_register_function(db, "cloudsync_is_enabled", dbsync_is_enabled, 1, pzErrMsg, ctx, NULL);
2009 759 : if (rc != SQLITE_OK) return rc;
2010 :
2011 759 : rc = dbsync_register_function(db, "cloudsync_cleanup", dbsync_cleanup, 1, pzErrMsg, ctx, NULL);
2012 759 : if (rc != SQLITE_OK) return rc;
2013 :
2014 759 : rc = dbsync_register_function(db, "cloudsync_terminate", dbsync_terminate, 0, pzErrMsg, ctx, NULL);
2015 759 : if (rc != SQLITE_OK) return rc;
2016 :
2017 759 : rc = dbsync_register_function(db, "cloudsync_set", dbsync_set, 2, pzErrMsg, ctx, NULL);
2018 759 : if (rc != SQLITE_OK) return rc;
2019 :
2020 759 : rc = dbsync_register_function(db, "cloudsync_set_table", dbsync_set_table, 3, pzErrMsg, ctx, NULL);
2021 759 : if (rc != SQLITE_OK) return rc;
2022 :
2023 759 : rc = dbsync_register_function(db, "cloudsync_set_filter", dbsync_set_filter, 2, pzErrMsg, ctx, NULL);
2024 759 : if (rc != SQLITE_OK) return rc;
2025 :
2026 759 : rc = dbsync_register_function(db, "cloudsync_clear_filter", dbsync_clear_filter, 1, pzErrMsg, ctx, NULL);
2027 759 : if (rc != SQLITE_OK) return rc;
2028 :
2029 759 : rc = dbsync_register_function(db, "cloudsync_set_schema", dbsync_set_schema, 1, pzErrMsg, ctx, NULL);
2030 759 : if (rc != SQLITE_OK) return rc;
2031 :
2032 759 : rc = dbsync_register_function(db, "cloudsync_schema", dbsync_schema, 0, pzErrMsg, ctx, NULL);
2033 759 : if (rc != SQLITE_OK) return rc;
2034 :
2035 759 : rc = dbsync_register_function(db, "cloudsync_table_schema", dbsync_table_schema, 1, pzErrMsg, ctx, NULL);
2036 759 : if (rc != SQLITE_OK) return rc;
2037 :
2038 759 : rc = dbsync_register_function(db, "cloudsync_set_column", dbsync_set_column, 4, pzErrMsg, ctx, NULL);
2039 759 : if (rc != SQLITE_OK) return rc;
2040 :
2041 759 : rc = dbsync_register_function(db, "cloudsync_siteid", dbsync_siteid, 0, pzErrMsg, ctx, NULL);
2042 759 : if (rc != SQLITE_OK) return rc;
2043 :
2044 759 : rc = dbsync_register_function(db, "cloudsync_db_version", dbsync_db_version, 0, pzErrMsg, ctx, NULL);
2045 759 : if (rc != SQLITE_OK) return rc;
2046 :
2047 759 : rc = dbsync_register_function(db, "cloudsync_db_version_next", dbsync_db_version_next, 0, pzErrMsg, ctx, NULL);
2048 759 : if (rc != SQLITE_OK) return rc;
2049 :
2050 759 : rc = dbsync_register_function(db, "cloudsync_db_version_next", dbsync_db_version_next, 1, pzErrMsg, ctx, NULL);
2051 759 : if (rc != SQLITE_OK) return rc;
2052 :
2053 759 : rc = dbsync_register_function(db, "cloudsync_begin_alter", dbsync_begin_alter, 1, pzErrMsg, ctx, NULL);
2054 759 : if (rc != SQLITE_OK) return rc;
2055 :
2056 759 : rc = dbsync_register_function(db, "cloudsync_commit_alter", dbsync_commit_alter, 1, pzErrMsg, ctx, NULL);
2057 759 : if (rc != SQLITE_OK) return rc;
2058 :
2059 759 : rc = dbsync_register_function(db, "cloudsync_uuid", dbsync_uuid, 0, pzErrMsg, ctx, NULL);
2060 759 : if (rc != SQLITE_OK) return rc;
2061 :
2062 759 : rc = dbsync_register_function(db, "cloudsync_uuid_text", dbsync_uuid_text, 1, pzErrMsg, ctx, NULL);
2063 759 : if (rc != SQLITE_OK) return rc;
2064 759 : rc = dbsync_register_function(db, "cloudsync_uuid_text", dbsync_uuid_text, 2, pzErrMsg, ctx, NULL);
2065 759 : if (rc != SQLITE_OK) return rc;
2066 759 : rc = dbsync_register_function(db, "cloudsync_uuid_blob", dbsync_uuid_blob, 1, pzErrMsg, ctx, NULL);
2067 759 : if (rc != SQLITE_OK) return rc;
2068 :
2069 : // PAYLOAD
2070 759 : rc = dbsync_register_aggregate(db, "cloudsync_payload_encode", dbsync_payload_encode_step, dbsync_payload_encode_final, -1, pzErrMsg, ctx, NULL);
2071 759 : if (rc != SQLITE_OK) return rc;
2072 :
2073 : // alias
2074 759 : rc = dbsync_register_function(db, "cloudsync_payload_decode", dbsync_payload_decode, -1, pzErrMsg, ctx, NULL);
2075 759 : if (rc != SQLITE_OK) return rc;
2076 759 : rc = dbsync_register_function(db, "cloudsync_payload_apply", dbsync_payload_decode, -1, pzErrMsg, ctx, NULL);
2077 759 : if (rc != SQLITE_OK) return rc;
2078 :
2079 759 : rc = sqlite3_create_module(db, "cloudsync_payload_chunks", &cloudsync_payload_chunks_module, (void *)ctx);
2080 759 : if (rc != SQLITE_OK) return rc;
2081 :
2082 759 : rc = dbsync_register_function(db, "cloudsync_payload_blob_checked", dbsync_payload_blob_checked, 5, pzErrMsg, ctx, NULL);
2083 759 : if (rc != SQLITE_OK) return rc;
2084 :
2085 : #ifdef CLOUDSYNC_DESKTOP_OS
2086 759 : rc = dbsync_register_function(db, "cloudsync_payload_save", dbsync_payload_save, 1, pzErrMsg, ctx, NULL);
2087 759 : if (rc != SQLITE_OK) return rc;
2088 :
2089 759 : rc = dbsync_register_function(db, "cloudsync_payload_load", dbsync_payload_load, 1, pzErrMsg, ctx, NULL);
2090 759 : if (rc != SQLITE_OK) return rc;
2091 : #endif
2092 :
2093 : // PRIVATE functions (used inside triggers — require SQLITE_INNOCUOUS)
2094 759 : rc = dbsync_register_trigger_function(db, "cloudsync_is_sync", dbsync_is_sync, 1, pzErrMsg, ctx, NULL);
2095 759 : if (rc != SQLITE_OK) return rc;
2096 :
2097 759 : rc = dbsync_register_trigger_function(db, "cloudsync_insert", dbsync_insert, -1, pzErrMsg, ctx, NULL);
2098 759 : if (rc != SQLITE_OK) return rc;
2099 :
2100 759 : rc = dbsync_register_trigger_aggregate(db, "cloudsync_update", dbsync_update_step, dbsync_update_final, 3, pzErrMsg, ctx, NULL);
2101 759 : if (rc != SQLITE_OK) return rc;
2102 :
2103 759 : rc = dbsync_register_trigger_function(db, "cloudsync_delete", dbsync_delete, -1, pzErrMsg, ctx, NULL);
2104 759 : if (rc != SQLITE_OK) return rc;
2105 :
2106 759 : rc = dbsync_register_function(db, "cloudsync_col_value", dbsync_col_value, 3, pzErrMsg, ctx, NULL);
2107 759 : if (rc != SQLITE_OK) return rc;
2108 :
2109 759 : rc = dbsync_register_pure_function(db, "cloudsync_pk_encode", dbsync_pk_encode, -1, pzErrMsg, ctx, NULL);
2110 759 : if (rc != SQLITE_OK) return rc;
2111 :
2112 759 : rc = dbsync_register_pure_function(db, "cloudsync_pk_decode", dbsync_pk_decode, 2, pzErrMsg, ctx, NULL);
2113 759 : if (rc != SQLITE_OK) return rc;
2114 :
2115 759 : rc = dbsync_register_function(db, "cloudsync_seq", dbsync_seq, 0, pzErrMsg, ctx, NULL);
2116 759 : if (rc != SQLITE_OK) return rc;
2117 :
2118 759 : rc = dbsync_register_function(db, "cloudsync_text_materialize", dbsync_text_materialize, -1, pzErrMsg, ctx, NULL);
2119 759 : if (rc != SQLITE_OK) return rc;
2120 :
2121 : // NETWORK LAYER
2122 : #ifndef CLOUDSYNC_OMIT_NETWORK
2123 16 : rc = cloudsync_network_register(db, pzErrMsg, ctx);
2124 16 : if (rc != SQLITE_OK) return rc;
2125 : #endif
2126 :
2127 759 : cloudsync_context *data = (cloudsync_context *)ctx;
2128 759 : sqlite3_commit_hook(db, cloudsync_commit_hook, ctx);
2129 759 : sqlite3_rollback_hook(db, cloudsync_rollback_hook, ctx);
2130 :
2131 : // register eponymous only changes virtual table
2132 759 : rc = cloudsync_vtab_register_changes (db, data);
2133 759 : if (rc != SQLITE_OK) {
2134 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("Error creating changes virtual table: %s", sqlite3_errmsg(db));
2135 0 : return rc;
2136 : }
2137 :
2138 : // load config, if exists
2139 759 : if (cloudsync_config_exists(data)) {
2140 6 : if (cloudsync_context_init(data) == NULL) {
2141 : // Do not free ctx here: it is already owned by the cloudsync_version
2142 : // function (registered above with cloudsync_context_free as its
2143 : // destructor). SQLite will release it when the connection is closed.
2144 : // Freeing it manually would cause a double-free on sqlite3_close.
2145 0 : if (pzErrMsg) *pzErrMsg = sqlite3_mprintf("An error occurred while trying to initialize context");
2146 0 : return SQLITE_ERROR;
2147 : }
2148 :
2149 : // update schema hash if upgrading from an older version
2150 6 : if (dbutils_settings_check_version(data, NULL) != 0) {
2151 0 : cloudsync_update_schema_hash(data);
2152 0 : }
2153 :
2154 : // make sure to update internal version to current version
2155 6 : dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_LIBVERSION, CLOUDSYNC_VERSION);
2156 6 : }
2157 :
2158 759 : return SQLITE_OK;
2159 761 : }
2160 :
2161 : // MARK: - Main Entrypoint -
2162 :
2163 761 : APIEXPORT int sqlite3_cloudsync_init (sqlite3 *db, char **pzErrMsg, const sqlite3_api_routines *pApi) {
2164 : DEBUG_FUNCTION("sqlite3_cloudsync_init");
2165 :
2166 : #ifndef SQLITE_CORE
2167 : SQLITE_EXTENSION_INIT2(pApi);
2168 : #endif
2169 :
2170 761 : return dbsync_register_functions(db, pzErrMsg);
2171 : }
|
