Line data Source code
1 : //
2 : // cloudsync.c
3 : // cloudsync
4 : //
5 : // Created by Marco Bambini on 16/05/24.
6 : //
7 :
8 : #include <inttypes.h>
9 : #include <stdbool.h>
10 : #include <limits.h>
11 : #include <stdint.h>
12 : #include <stdlib.h>
13 : #include <string.h>
14 : #include <assert.h>
15 : #include <stdio.h>
16 : #include <errno.h>
17 : #include <math.h>
18 : #include <time.h>
19 :
20 : #include "cloudsync.h"
21 : #include "lz4.h"
22 : #include "pk.h"
23 : #include "sql.h"
24 : #include "utils.h"
25 : #include "dbutils.h"
26 : #include "block.h"
27 :
28 : #ifdef _WIN32
29 : #include <winsock2.h>
30 : #include <ws2tcpip.h>
31 : #else
32 : #include <arpa/inet.h> // for htonl, htons, ntohl, ntohs
33 : #include <netinet/in.h> // for struct sockaddr_in, INADDR_ANY, etc. (if needed)
34 : #endif
35 :
36 : #ifndef htonll
37 : #if __BIG_ENDIAN__
38 : #define htonll(x) (x)
39 : #define ntohll(x) (x)
40 : #else
41 : #ifndef htobe64
42 : #define htonll(x) ((uint64_t)htonl((x) & 0xFFFFFFFF) << 32 | (uint64_t)htonl((x) >> 32))
43 : #define ntohll(x) ((uint64_t)ntohl((x) & 0xFFFFFFFF) << 32 | (uint64_t)ntohl((x) >> 32))
44 : #else
45 : #define htonll(x) htobe64(x)
46 : #define ntohll(x) be64toh(x)
47 : #endif
48 : #endif
49 : #endif
50 :
51 : #define CLOUDSYNC_INIT_NTABLES 64
52 : #define CLOUDSYNC_MIN_DB_VERSION 0
53 :
54 : #define CLOUDSYNC_PAYLOAD_MINBUF_SIZE (512*1024)
55 : #define CLOUDSYNC_PAYLOAD_SIGNATURE 0x434C5359 /* 'C','L','S','Y' */
56 : #define CLOUDSYNC_PAYLOAD_VERSION_ORIGNAL 1
57 : #define CLOUDSYNC_PAYLOAD_VERSION_1 CLOUDSYNC_PAYLOAD_VERSION_ORIGNAL
58 : #define CLOUDSYNC_PAYLOAD_VERSION_2 2
59 : #define CLOUDSYNC_PAYLOAD_VERSION_3 3
60 : #define CLOUDSYNC_PAYLOAD_VERSION_LATEST CLOUDSYNC_PAYLOAD_VERSION_2
61 : #define CLOUDSYNC_PAYLOAD_MIN_VERSION_WITH_CHECKSUM CLOUDSYNC_PAYLOAD_VERSION_2
62 : #define CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX "__cloudsync_frag_v1__:"
63 : #define CLOUDSYNC_PAYLOAD_FRAGMENT_STALE_SECONDS (24*60*60)
64 : #define CLOUDSYNC_PAYLOAD_FRAGMENT_CLEANUP_MIN_INTERVAL (60)
65 :
66 : #ifndef MAX
67 : #define MAX(a, b) (((a)>(b))?(a):(b))
68 : #endif
69 :
70 : #define DEBUG_DBERROR(_rc, _fn, _data) do {if (_rc != DBRES_OK) printf("Error in %s: %s\n", _fn, database_errmsg(_data));} while (0)
71 :
72 : typedef enum {
73 : CLOUDSYNC_PK_INDEX_TBL = 0,
74 : CLOUDSYNC_PK_INDEX_PK = 1,
75 : CLOUDSYNC_PK_INDEX_COLNAME = 2,
76 : CLOUDSYNC_PK_INDEX_COLVALUE = 3,
77 : CLOUDSYNC_PK_INDEX_COLVERSION = 4,
78 : CLOUDSYNC_PK_INDEX_DBVERSION = 5,
79 : CLOUDSYNC_PK_INDEX_SITEID = 6,
80 : CLOUDSYNC_PK_INDEX_CL = 7,
81 : CLOUDSYNC_PK_INDEX_SEQ = 8
82 : } CLOUDSYNC_PK_INDEX;
83 :
84 : typedef enum {
85 : DBVM_VALUE_ERROR = -1,
86 : DBVM_VALUE_UNCHANGED = 0,
87 : DBVM_VALUE_CHANGED = 1,
88 : } DBVM_VALUE;
89 :
90 : #define SYNCBIT_SET(_data) _data->insync = 1
91 : #define SYNCBIT_RESET(_data) _data->insync = 0
92 :
93 : // MARK: - Deferred column-batch merge -
94 :
95 : typedef struct {
96 : const char *col_name; // pointer into table_context->col_name[idx] (stable)
97 : dbvalue_t *col_value; // duplicated via database_value_dup (owned)
98 : int64_t col_version;
99 : int64_t db_version;
100 : uint8_t site_id[UUID_LEN];
101 : int site_id_len;
102 : int64_t seq;
103 : } merge_pending_entry;
104 :
105 : typedef struct {
106 : cloudsync_table_context *table;
107 : char *pk; // malloc'd copy, freed on flush
108 : int pk_len;
109 : int64_t cl;
110 : bool sentinel_pending;
111 : bool row_exists; // true when the PK already exists locally
112 : int count;
113 : int capacity;
114 : merge_pending_entry *entries;
115 :
116 : // Statement cache — reuse the prepared statement when the column
117 : // combination and row_exists flag match between consecutive PK flushes.
118 : dbvm_t *cached_vm;
119 : bool cached_row_exists;
120 : int cached_col_count;
121 : const char **cached_col_names; // array of pointers into table_context (not owned)
122 : } merge_pending_batch;
123 :
124 : // MARK: -
125 :
126 : struct cloudsync_pk_decode_bind_context {
127 : dbvm_t *vm;
128 : char *tbl;
129 : int64_t tbl_len;
130 : const void *pk;
131 : int64_t pk_len;
132 : char *col_name;
133 : int64_t col_name_len;
134 : int64_t col_version;
135 : int64_t db_version;
136 : const void *site_id;
137 : int64_t site_id_len;
138 : int64_t cl;
139 : int64_t seq;
140 : };
141 :
142 : struct cloudsync_context {
143 : void *db;
144 : char errmsg[1024];
145 : int errcode;
146 :
147 : char *libversion;
148 : uint8_t site_id[UUID_LEN];
149 : int insync;
150 : int debug;
151 : bool merge_equal_values;
152 : void *aux_data;
153 :
154 : // stmts and context values
155 : dbvm_t *schema_version_stmt;
156 : dbvm_t *data_version_stmt;
157 : dbvm_t *db_version_stmt;
158 : dbvm_t *getset_siteid_stmt;
159 : int data_version;
160 : int schema_version;
161 : uint64_t schema_hash;
162 :
163 : // set at transaction start and reset on commit/rollback
164 : int64_t db_version;
165 : // version the DB would have if the transaction committed now
166 : int64_t pending_db_version;
167 : // used to set an order inside each transaction
168 : int seq;
169 :
170 : // wall-clock (time()) of the last stale v3-fragment GC; throttles the GC so
171 : // it does not run a full table scan on every applied fragment (0 = never run)
172 : int64_t last_fragment_cleanup;
173 :
174 : // optional schema_name to be set in the cloudsync_table_context
175 : char *current_schema;
176 :
177 : // augmented tables are stored in-memory so we do not need to retrieve information about
178 : // col_names and cid from the disk each time a write statement is performed
179 : // we do also not need to use an hash map here because for few tables the direct
180 : // in-memory comparison with table name is faster
181 : cloudsync_table_context **tables; // dense vector: [0..tables_count-1] are valid
182 : int tables_count; // size
183 : int tables_cap; // capacity
184 :
185 : int skip_decode_idx; // -1 in sqlite, col_value index in postgresql
186 :
187 : // deferred column-batch merge (active during payload_apply)
188 : merge_pending_batch *pending_batch;
189 :
190 : // last (db_version, seq) successfully applied during the current
191 : // cloudsync_payload_apply call; used to resolve the
192 : // CLOUDSYNC_CHECKPOINT_LAST_APPLIED receive-checkpoint mode (-1 = none yet).
193 : int64_t apply_last_db_version;
194 : int64_t apply_last_seq;
195 : };
196 :
197 : struct cloudsync_table_context {
198 : table_algo algo; // CRDT algoritm associated to the table
199 : char *name; // table name
200 : char *schema; // table schema
201 : char *meta_ref; // schema-qualified meta table name (e.g. "schema"."name_cloudsync")
202 : char *base_ref; // schema-qualified base table name (e.g. "schema"."name")
203 : char **col_name; // array of column names
204 : dbvm_t **col_merge_stmt; // array of merge insert stmt (indexed by col_name)
205 : dbvm_t **col_value_stmt; // array of column value stmt (indexed by col_name)
206 : int *col_id; // array of column id
207 : col_algo_t *col_algo; // per-column algorithm (normal or block)
208 : char **col_delimiter; // per-column delimiter for block splitting (NULL = default "\n")
209 : bool has_block_cols; // quick check: does this table have any block columns?
210 : dbvm_t *block_value_read_stmt; // SELECT col_value FROM blocks table
211 : dbvm_t *block_value_write_stmt; // INSERT OR REPLACE into blocks table
212 : dbvm_t *block_value_delete_stmt; // DELETE from blocks table
213 : dbvm_t *block_list_stmt; // SELECT block entries for materialization
214 : char *blocks_ref; // schema-qualified blocks table name
215 : int ncols; // number of non primary key cols
216 : int npks; // number of primary key cols
217 : bool enabled; // flag to check if a table is enabled or disabled
218 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
219 : bool rowid_only; // a table with no primary keys other than the implicit rowid
220 : #endif
221 :
222 : char **pk_name; // array of primary key names
223 :
224 : // precompiled statements
225 : dbvm_t *meta_pkexists_stmt; // check if a primary key already exist in the augmented table
226 : dbvm_t *meta_sentinel_update_stmt; // update a local sentinel row
227 : dbvm_t *meta_sentinel_insert_stmt; // insert a local sentinel row
228 : dbvm_t *meta_row_insert_update_stmt; // insert/update a local row
229 : dbvm_t *meta_row_drop_stmt; // delete rows from meta
230 : dbvm_t *meta_update_move_stmt; // update rows in meta when pk changes
231 : dbvm_t *meta_local_cl_stmt; // compute local cl value
232 : dbvm_t *meta_winner_clock_stmt; // get the rowid of the last inserted/updated row in the meta table
233 : dbvm_t *meta_merge_delete_drop;
234 : dbvm_t *meta_zero_clock_stmt;
235 : dbvm_t *meta_col_version_stmt;
236 : dbvm_t *meta_site_id_stmt;
237 :
238 : dbvm_t *real_col_values_stmt; // retrieve all column values based on pk
239 : dbvm_t *real_merge_delete_stmt;
240 : dbvm_t *real_merge_sentinel_stmt;
241 :
242 : bool is_altering; // flag to track if a table alteration is in progress
243 :
244 : // context
245 : cloudsync_context *context;
246 : };
247 :
248 : struct cloudsync_payload_context {
249 : char *buffer;
250 : size_t bsize;
251 : size_t balloc;
252 : size_t bused;
253 : uint64_t nrows;
254 : uint16_t ncols;
255 : uint8_t version;
256 : };
257 :
258 : #ifdef _MSC_VER
259 : #pragma pack(push, 1) // For MSVC: pack struct with 1-byte alignment
260 : #define PACKED
261 : #else
262 : #define PACKED __attribute__((__packed__))
263 : #endif
264 :
265 : typedef struct PACKED {
266 : uint32_t signature; // 'CLSY'
267 : uint8_t version; // protocol version
268 : uint8_t libversion[3]; // major.minor.patch
269 : uint32_t expanded_size;
270 : uint16_t ncols;
271 : uint32_t nrows;
272 : uint64_t schema_hash;
273 : uint8_t checksum[6]; // 48 bits checksum (to ensure struct is 32 bytes)
274 : } cloudsync_payload_header;
275 :
276 : #ifdef _MSC_VER
277 : #pragma pack(pop)
278 : #endif
279 :
280 : #if CLOUDSYNC_UNITTEST
281 : bool force_uncompressed_blob = false;
282 : #define CHECK_FORCE_UNCOMPRESSED_BUFFER() if (force_uncompressed_blob) use_uncompressed_buffer = true
283 : #else
284 : #define CHECK_FORCE_UNCOMPRESSED_BUFFER()
285 : #endif
286 :
287 : // Internal prototypes
288 : int local_mark_insert_or_update_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const char *col_name, int64_t db_version, int seq);
289 :
290 : // MARK: - CRDT algos -
291 :
292 660 : table_algo cloudsync_algo_from_name (const char *algo_name) {
293 660 : if (algo_name == NULL) return table_algo_none;
294 :
295 660 : if ((strcasecmp(algo_name, "CausalLengthSet") == 0) || (strcasecmp(algo_name, "cls") == 0)) return table_algo_crdt_cls;
296 284 : if ((strcasecmp(algo_name, "GrowOnlySet") == 0) || (strcasecmp(algo_name, "gos") == 0)) return table_algo_crdt_gos;
297 277 : if ((strcasecmp(algo_name, "DeleteWinsSet") == 0) || (strcasecmp(algo_name, "dws") == 0)) return table_algo_crdt_dws;
298 276 : if ((strcasecmp(algo_name, "AddWinsSet") == 0) || (strcasecmp(algo_name, "aws") == 0)) return table_algo_crdt_aws;
299 :
300 : // if nothing is found
301 275 : return table_algo_none;
302 660 : }
303 :
304 110 : const char *cloudsync_algo_name (table_algo algo) {
305 110 : switch (algo) {
306 105 : case table_algo_crdt_cls: return "cls";
307 1 : case table_algo_crdt_gos: return "gos";
308 1 : case table_algo_crdt_dws: return "dws";
309 1 : case table_algo_crdt_aws: return "aws";
310 1 : case table_algo_none: return NULL;
311 : }
312 1 : return NULL;
313 110 : }
314 :
315 : // MARK: - DBVM Utils -
316 :
317 35484 : DBVM_VALUE dbvm_execute (dbvm_t *stmt, cloudsync_context *data) {
318 35484 : if (!stmt) return DBVM_VALUE_ERROR;
319 :
320 35482 : int rc = databasevm_step(stmt);
321 35482 : if (rc != DBRES_ROW && rc != DBRES_DONE) {
322 2 : if (data) DEBUG_DBERROR(rc, "stmt_execute", data);
323 2 : databasevm_reset(stmt);
324 2 : return DBVM_VALUE_ERROR;
325 : }
326 :
327 35480 : DBVM_VALUE result = DBVM_VALUE_CHANGED;
328 35480 : if (stmt == data->data_version_stmt) {
329 33000 : int version = (int)database_column_int(stmt, 0);
330 33000 : if (version != data->data_version) {
331 243 : data->data_version = version;
332 243 : } else {
333 32757 : result = DBVM_VALUE_UNCHANGED;
334 : }
335 35480 : } else if (stmt == data->schema_version_stmt) {
336 1240 : int version = (int)database_column_int(stmt, 0);
337 1240 : if (version > data->schema_version) {
338 398 : data->schema_version = version;
339 398 : } else {
340 842 : result = DBVM_VALUE_UNCHANGED;
341 : }
342 :
343 2480 : } else if (stmt == data->db_version_stmt) {
344 1240 : data->db_version = (rc == DBRES_DONE) ? CLOUDSYNC_MIN_DB_VERSION : database_column_int(stmt, 0);
345 1240 : }
346 :
347 35480 : databasevm_reset(stmt);
348 35480 : return result;
349 35484 : }
350 :
351 4857 : int dbvm_count (dbvm_t *stmt, const char *value, size_t len, int type) {
352 4857 : int result = -1;
353 4857 : int rc = DBRES_OK;
354 :
355 4857 : if (value) {
356 4856 : rc = (type == DBTYPE_TEXT) ? databasevm_bind_text(stmt, 1, value, (int)len) : databasevm_bind_blob(stmt, 1, value, len);
357 4856 : if (rc != DBRES_OK) goto cleanup;
358 4856 : }
359 :
360 4857 : rc = databasevm_step(stmt);
361 9714 : if (rc == DBRES_DONE) {
362 1 : result = 0;
363 1 : rc = DBRES_OK;
364 4857 : } else if (rc == DBRES_ROW) {
365 4856 : result = (int)database_column_int(stmt, 0);
366 4856 : rc = DBRES_OK;
367 4856 : }
368 :
369 : cleanup:
370 4857 : databasevm_reset(stmt);
371 4857 : return result;
372 : }
373 :
374 269278 : void dbvm_reset (dbvm_t *stmt) {
375 269278 : if (!stmt) return;
376 246147 : databasevm_clear_bindings(stmt);
377 246147 : databasevm_reset(stmt);
378 269278 : }
379 :
380 : // MARK: - Database Version -
381 :
382 709 : int cloudsync_dbversion_build_query (cloudsync_context *data, char **sql_out) {
383 : // this function must be manually called each time tables changes
384 : // because the query plan changes too and it must be re-prepared
385 : // unfortunately there is no other way
386 :
387 : // we need to execute a query like:
388 : /*
389 : SELECT max(version) as version FROM (
390 : SELECT max(db_version) as version FROM "table1_cloudsync"
391 : UNION ALL
392 : SELECT max(db_version) as version FROM "table2_cloudsync"
393 : UNION ALL
394 : SELECT max(db_version) as version FROM "table3_cloudsync"
395 : UNION
396 : SELECT value as version FROM cloudsync_settings WHERE key = 'pre_alter_dbversion'
397 : )
398 : */
399 :
400 : // the good news is that the query can be computed in SQLite without the need to do any extra computation from the host language
401 :
402 709 : *sql_out = NULL;
403 709 : return database_select_text(data, SQL_DBVERSION_BUILD_QUERY, sql_out);
404 : }
405 :
406 950 : int cloudsync_dbversion_rebuild (cloudsync_context *data) {
407 950 : if (data->db_version_stmt) {
408 461 : databasevm_finalize(data->db_version_stmt);
409 461 : data->db_version_stmt = NULL;
410 461 : }
411 :
412 950 : int64_t count = dbutils_table_settings_count_tables(data);
413 950 : if (count == 0) return DBRES_OK;
414 709 : else if (count == -1) return cloudsync_set_dberror(data);
415 :
416 709 : char *sql = NULL;
417 709 : int rc = cloudsync_dbversion_build_query(data, &sql);
418 709 : if (rc != DBRES_OK) return cloudsync_set_dberror(data);
419 :
420 : // A NULL SQL with rc == OK means the generator produced a NULL row:
421 : // sqlite_master has no *_cloudsync meta-tables (for example, the user
422 : // dropped the base table and its meta-table without calling
423 : // cloudsync_cleanup, leaving stale cloudsync_table_settings rows).
424 : // Treat this the same as count == 0: no prepared statement, db_version
425 : // stays at the minimum and will be rebuilt on the next cloudsync_init.
426 : // Genuine errors from database_select_text are handled above.
427 708 : if (!sql) return DBRES_OK;
428 : DEBUG_SQL("db_version_stmt: %s", sql);
429 :
430 707 : rc = databasevm_prepare(data, sql, (void **)&data->db_version_stmt, DBFLAG_PERSISTENT);
431 : DEBUG_STMT("db_version_stmt %p", data->db_version_stmt);
432 707 : cloudsync_memory_free(sql);
433 707 : return rc;
434 950 : }
435 :
436 1240 : int cloudsync_dbversion_rerun (cloudsync_context *data) {
437 1240 : DBVM_VALUE schema_changed = dbvm_execute(data->schema_version_stmt, data);
438 1240 : if (schema_changed == DBVM_VALUE_ERROR) return -1;
439 :
440 1240 : if (schema_changed == DBVM_VALUE_CHANGED) {
441 398 : int rc = cloudsync_dbversion_rebuild(data);
442 398 : if (rc != DBRES_OK) return -1;
443 398 : }
444 :
445 1240 : if (!data->db_version_stmt) {
446 0 : data->db_version = CLOUDSYNC_MIN_DB_VERSION;
447 0 : return 0;
448 : }
449 :
450 1240 : DBVM_VALUE rc = dbvm_execute(data->db_version_stmt, data);
451 1240 : if (rc == DBVM_VALUE_ERROR) return -1;
452 1240 : return 0;
453 1240 : }
454 :
455 33002 : int cloudsync_dbversion_check_uptodate (cloudsync_context *data) {
456 : // perform a PRAGMA data_version to check if some other process write any data
457 33002 : DBVM_VALUE rc = dbvm_execute(data->data_version_stmt, data);
458 33002 : if (rc == DBVM_VALUE_ERROR) return -1;
459 :
460 : // db_version is already set and there is no need to update it
461 33000 : if (data->db_version != CLOUDSYNC_VALUE_NOTSET && rc == DBVM_VALUE_UNCHANGED) return 0;
462 :
463 1240 : return cloudsync_dbversion_rerun(data);
464 33002 : }
465 :
466 32976 : int64_t cloudsync_dbversion_next (cloudsync_context *data, int64_t merging_version) {
467 32976 : int rc = cloudsync_dbversion_check_uptodate(data);
468 32976 : if (rc != DBRES_OK) return -1;
469 :
470 32975 : int64_t result = data->db_version + 1;
471 32975 : if (result < data->pending_db_version) result = data->pending_db_version;
472 32975 : if (merging_version != CLOUDSYNC_VALUE_NOTSET && result < merging_version) result = merging_version;
473 32975 : data->pending_db_version = result;
474 :
475 32975 : return result;
476 32976 : }
477 :
478 : // MARK: - PK Context -
479 :
480 0 : char *cloudsync_pk_context_tbl (cloudsync_pk_decode_bind_context *ctx, int64_t *tbl_len) {
481 0 : *tbl_len = ctx->tbl_len;
482 0 : return ctx->tbl;
483 : }
484 :
485 0 : void *cloudsync_pk_context_pk (cloudsync_pk_decode_bind_context *ctx, int64_t *pk_len) {
486 0 : *pk_len = ctx->pk_len;
487 0 : return (void *)ctx->pk;
488 : }
489 :
490 0 : char *cloudsync_pk_context_colname (cloudsync_pk_decode_bind_context *ctx, int64_t *colname_len) {
491 0 : *colname_len = ctx->col_name_len;
492 0 : return ctx->col_name;
493 : }
494 :
495 0 : int64_t cloudsync_pk_context_cl (cloudsync_pk_decode_bind_context *ctx) {
496 0 : return ctx->cl;
497 : }
498 :
499 0 : int64_t cloudsync_pk_context_dbversion (cloudsync_pk_decode_bind_context *ctx) {
500 0 : return ctx->db_version;
501 : }
502 :
503 : // MARK: - CloudSync Context -
504 :
505 15497 : int cloudsync_insync (cloudsync_context *data) {
506 15497 : return data->insync;
507 : }
508 :
509 838 : void *cloudsync_siteid (cloudsync_context *data) {
510 838 : return (void *)data->site_id;
511 : }
512 :
513 2 : void cloudsync_reset_siteid (cloudsync_context *data) {
514 2 : memset(data->site_id, 0, sizeof(uint8_t) * UUID_LEN);
515 2 : }
516 :
517 249 : int cloudsync_load_siteid (cloudsync_context *data) {
518 : // check if site_id was already loaded
519 249 : if (data->site_id[0] != 0) return DBRES_OK;
520 :
521 : // load site_id
522 247 : char *buffer = NULL;
523 247 : int64_t size = 0;
524 247 : int rc = database_select_blob(data, SQL_SITEID_SELECT_ROWID0, &buffer, &size);
525 247 : if (rc != DBRES_OK) return rc;
526 247 : if (!buffer || size != UUID_LEN) {
527 0 : if (buffer) cloudsync_memory_free(buffer);
528 0 : return cloudsync_set_error(data, "Unable to retrieve siteid", DBRES_MISUSE);
529 : }
530 :
531 247 : memcpy(data->site_id, buffer, UUID_LEN);
532 247 : cloudsync_memory_free(buffer);
533 :
534 247 : return DBRES_OK;
535 249 : }
536 :
537 2 : int64_t cloudsync_dbversion (cloudsync_context *data) {
538 2 : return data->db_version;
539 : }
540 :
541 13686 : int cloudsync_bumpseq (cloudsync_context *data) {
542 13686 : int value = data->seq;
543 13686 : data->seq += 1;
544 13686 : return value;
545 : }
546 :
547 304 : void cloudsync_update_schema_hash (cloudsync_context *data) {
548 304 : database_update_schema_hash(data, &data->schema_hash);
549 304 : }
550 :
551 67063 : void *cloudsync_db (cloudsync_context *data) {
552 67063 : return data->db;
553 : }
554 :
555 551 : int cloudsync_add_dbvms (cloudsync_context *data) {
556 : DEBUG_DBFUNCTION("cloudsync_add_stmts");
557 :
558 551 : if (data->data_version_stmt == NULL) {
559 247 : int rc = databasevm_prepare(data, SQL_DATA_VERSION, (void **)&data->data_version_stmt, DBFLAG_PERSISTENT);
560 : DEBUG_STMT("data_version_stmt %p", data->data_version_stmt);
561 247 : if (rc != DBRES_OK) return rc;
562 : DEBUG_SQL("data_version_stmt: %s", SQL_DATA_VERSION);
563 247 : }
564 :
565 551 : if (data->schema_version_stmt == NULL) {
566 247 : int rc = databasevm_prepare(data, SQL_SCHEMA_VERSION, (void **)&data->schema_version_stmt, DBFLAG_PERSISTENT);
567 : DEBUG_STMT("schema_version_stmt %p", data->schema_version_stmt);
568 247 : if (rc != DBRES_OK) return rc;
569 : DEBUG_SQL("schema_version_stmt: %s", SQL_SCHEMA_VERSION);
570 247 : }
571 :
572 551 : if (data->getset_siteid_stmt == NULL) {
573 : // get and set index of the site_id
574 : // in SQLite, we can’t directly combine an INSERT and a SELECT to both insert a row and return an identifier (rowid) in a single statement,
575 : // however, we can use a workaround by leveraging the INSERT statement with ON CONFLICT DO UPDATE and then combining it with RETURNING rowid
576 247 : int rc = databasevm_prepare(data, SQL_SITEID_GETSET_ROWID_BY_SITEID, (void **)&data->getset_siteid_stmt, DBFLAG_PERSISTENT);
577 : DEBUG_STMT("getset_siteid_stmt %p", data->getset_siteid_stmt);
578 247 : if (rc != DBRES_OK) return rc;
579 : DEBUG_SQL("getset_siteid_stmt: %s", SQL_SITEID_GETSET_ROWID_BY_SITEID);
580 247 : }
581 :
582 551 : return cloudsync_dbversion_rebuild(data);
583 551 : }
584 :
585 23 : int cloudsync_set_error (cloudsync_context *data, const char *err_user, int err_code) {
586 : // force err_code to be something different than OK
587 23 : if (err_code == DBRES_OK) err_code = database_errcode(data);
588 23 : if (err_code == DBRES_OK) err_code = DBRES_ERROR;
589 :
590 : // compute a meaningful error message
591 23 : if (err_user == NULL) {
592 6 : snprintf(data->errmsg, sizeof(data->errmsg), "%s", database_errmsg(data));
593 6 : } else {
594 17 : const char *db_error = database_errmsg(data);
595 : char db_error_copy[sizeof(data->errmsg)];
596 17 : int rc = database_errcode(data);
597 17 : if (rc == DBRES_OK) {
598 17 : snprintf(data->errmsg, sizeof(data->errmsg), "%s", err_user);
599 17 : } else {
600 0 : if (db_error == data->errmsg) {
601 0 : snprintf(db_error_copy, sizeof(db_error_copy), "%s", db_error);
602 0 : db_error = db_error_copy;
603 0 : }
604 0 : snprintf(data->errmsg, sizeof(data->errmsg), "%s (%s)", err_user, db_error);
605 : }
606 : }
607 :
608 23 : data->errcode = err_code;
609 23 : return err_code;
610 : }
611 :
612 6 : int cloudsync_set_dberror (cloudsync_context *data) {
613 6 : return cloudsync_set_error(data, NULL, DBRES_OK);
614 : }
615 :
616 14 : const char *cloudsync_errmsg (cloudsync_context *data) {
617 14 : return data->errmsg;
618 : }
619 :
620 4 : int cloudsync_errcode (cloudsync_context *data) {
621 4 : return data->errcode;
622 : }
623 :
624 2 : void cloudsync_reset_error (cloudsync_context *data) {
625 2 : data->errmsg[0] = 0;
626 2 : data->errcode = DBRES_OK;
627 2 : }
628 :
629 3 : void *cloudsync_auxdata (cloudsync_context *data) {
630 3 : return data->aux_data;
631 : }
632 :
633 2 : void cloudsync_set_auxdata (cloudsync_context *data, void *xdata) {
634 2 : data->aux_data = xdata;
635 2 : }
636 :
637 6 : void cloudsync_set_schema (cloudsync_context *data, const char *schema) {
638 6 : if (data->current_schema && schema && strcmp(data->current_schema, schema) == 0) return;
639 5 : if (data->current_schema) cloudsync_memory_free(data->current_schema);
640 5 : data->current_schema = NULL;
641 5 : if (schema) data->current_schema = cloudsync_string_dup_lowercase(schema);
642 6 : }
643 :
644 1383 : const char *cloudsync_schema (cloudsync_context *data) {
645 1383 : return data->current_schema;
646 : }
647 :
648 4 : const char *cloudsync_table_schema (cloudsync_context *data, const char *table_name) {
649 4 : cloudsync_table_context *table = table_lookup(data, table_name);
650 4 : if (!table) return NULL;
651 :
652 1 : return table->schema;
653 4 : }
654 :
655 : // MARK: - Table Utils -
656 :
657 69 : void table_pknames_free (char **names, int nrows) {
658 69 : if (!names) return;
659 132 : for (int i = 0; i < nrows; ++i) {cloudsync_memory_free(names[i]);}
660 46 : cloudsync_memory_free(names);
661 69 : }
662 :
663 304 : char *table_build_mergedelete_sql (cloudsync_table_context *table) {
664 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
665 : if (table->rowid_only) {
666 : char *sql = memory_mprintf(SQL_DELETE_ROW_BY_ROWID, table->name);
667 : return sql;
668 : }
669 : #endif
670 :
671 304 : return sql_build_delete_by_pk(table->context, table->name, table->schema);
672 : }
673 :
674 1423 : char *table_build_mergeinsert_sql (cloudsync_table_context *table, const char *colname) {
675 1423 : char *sql = NULL;
676 :
677 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
678 : if (table->rowid_only) {
679 : if (colname == NULL) {
680 : // INSERT OR IGNORE INTO customers (first_name,last_name) VALUES (?,?);
681 : sql = memory_mprintf(SQL_INSERT_ROWID_IGNORE, table->name);
682 : } else {
683 : // INSERT INTO customers (first_name,last_name,age) VALUES (?,?,?) ON CONFLICT DO UPDATE SET age=?;
684 : sql = memory_mprintf(SQL_UPSERT_ROWID_AND_COL_BY_ROWID, table->name, colname, colname);
685 : }
686 : return sql;
687 : }
688 : #endif
689 :
690 1423 : if (colname == NULL) {
691 : // is sentinel insert
692 304 : sql = sql_build_insert_pk_ignore(table->context, table->name, table->schema);
693 304 : } else {
694 1119 : sql = sql_build_upsert_pk_and_col(table->context, table->name, colname, table->schema);
695 : }
696 1423 : return sql;
697 : }
698 :
699 1118 : char *table_build_value_sql (cloudsync_table_context *table, const char *colname) {
700 : #if !CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
701 : if (table->rowid_only) {
702 : char *colnamequote = "\"";
703 : char *sql = memory_mprintf(SQL_SELECT_COLS_BY_ROWID_FMT, colnamequote, colname, colnamequote, table->name);
704 : return sql;
705 : }
706 : #endif
707 :
708 : // SELECT age FROM customers WHERE first_name=? AND last_name=?;
709 1118 : return sql_build_select_cols_by_pk(table->context, table->name, colname, table->schema);
710 : }
711 :
712 304 : cloudsync_table_context *table_create (cloudsync_context *data, const char *name, table_algo algo) {
713 : DEBUG_DBFUNCTION("table_create %s", name);
714 :
715 304 : cloudsync_table_context *table = (cloudsync_table_context *)cloudsync_memory_zeroalloc(sizeof(cloudsync_table_context));
716 304 : if (!table) return NULL;
717 :
718 304 : table->context = data;
719 304 : table->algo = algo;
720 304 : table->name = cloudsync_string_dup_lowercase(name);
721 :
722 : // Detect schema from metadata table location. If metadata table doesn't
723 : // exist yet (during initialization), fall back to cloudsync_schema() which
724 : // returns the explicitly set schema or current_schema().
725 304 : table->schema = database_table_schema(name);
726 304 : if (!table->schema) {
727 304 : const char *fallback_schema = cloudsync_schema(data);
728 304 : if (fallback_schema) {
729 0 : table->schema = cloudsync_string_dup(fallback_schema);
730 0 : }
731 304 : }
732 :
733 304 : if (!table->name) {
734 0 : cloudsync_memory_free(table);
735 0 : return NULL;
736 : }
737 304 : table->meta_ref = database_build_meta_ref(table->schema, table->name);
738 304 : table->base_ref = database_build_base_ref(table->schema, table->name);
739 304 : table->enabled = true;
740 :
741 304 : return table;
742 304 : }
743 :
744 304 : void table_free (cloudsync_table_context *table) {
745 : DEBUG_DBFUNCTION("table_free %s", (table) ? (table->name) : "NULL");
746 304 : if (!table) return;
747 :
748 304 : if (table->col_name) {
749 1383 : for (int i=0; i<table->ncols; ++i) {
750 1118 : cloudsync_memory_free(table->col_name[i]);
751 1118 : }
752 265 : cloudsync_memory_free(table->col_name);
753 265 : }
754 304 : if (table->col_merge_stmt) {
755 1383 : for (int i=0; i<table->ncols; ++i) {
756 1118 : databasevm_finalize(table->col_merge_stmt[i]);
757 1118 : }
758 265 : cloudsync_memory_free(table->col_merge_stmt);
759 265 : }
760 304 : if (table->col_value_stmt) {
761 1383 : for (int i=0; i<table->ncols; ++i) {
762 1118 : databasevm_finalize(table->col_value_stmt[i]);
763 1118 : }
764 265 : cloudsync_memory_free(table->col_value_stmt);
765 265 : }
766 304 : if (table->col_id) {
767 265 : cloudsync_memory_free(table->col_id);
768 265 : }
769 304 : if (table->col_algo) {
770 265 : cloudsync_memory_free(table->col_algo);
771 265 : }
772 304 : if (table->col_delimiter) {
773 1383 : for (int i=0; i<table->ncols; ++i) {
774 1118 : if (table->col_delimiter[i]) cloudsync_memory_free(table->col_delimiter[i]);
775 1118 : }
776 265 : cloudsync_memory_free(table->col_delimiter);
777 265 : }
778 :
779 304 : if (table->block_value_read_stmt) databasevm_finalize(table->block_value_read_stmt);
780 304 : if (table->block_value_write_stmt) databasevm_finalize(table->block_value_write_stmt);
781 304 : if (table->block_value_delete_stmt) databasevm_finalize(table->block_value_delete_stmt);
782 304 : if (table->block_list_stmt) databasevm_finalize(table->block_list_stmt);
783 304 : if (table->blocks_ref) cloudsync_memory_free(table->blocks_ref);
784 :
785 304 : if (table->name) cloudsync_memory_free(table->name);
786 304 : if (table->schema) cloudsync_memory_free(table->schema);
787 304 : if (table->meta_ref) cloudsync_memory_free(table->meta_ref);
788 304 : if (table->base_ref) cloudsync_memory_free(table->base_ref);
789 304 : if (table->pk_name) table_pknames_free(table->pk_name, table->npks);
790 304 : if (table->meta_pkexists_stmt) databasevm_finalize(table->meta_pkexists_stmt);
791 304 : if (table->meta_sentinel_update_stmt) databasevm_finalize(table->meta_sentinel_update_stmt);
792 304 : if (table->meta_sentinel_insert_stmt) databasevm_finalize(table->meta_sentinel_insert_stmt);
793 304 : if (table->meta_row_insert_update_stmt) databasevm_finalize(table->meta_row_insert_update_stmt);
794 304 : if (table->meta_row_drop_stmt) databasevm_finalize(table->meta_row_drop_stmt);
795 304 : if (table->meta_update_move_stmt) databasevm_finalize(table->meta_update_move_stmt);
796 304 : if (table->meta_local_cl_stmt) databasevm_finalize(table->meta_local_cl_stmt);
797 304 : if (table->meta_winner_clock_stmt) databasevm_finalize(table->meta_winner_clock_stmt);
798 304 : if (table->meta_merge_delete_drop) databasevm_finalize(table->meta_merge_delete_drop);
799 304 : if (table->meta_zero_clock_stmt) databasevm_finalize(table->meta_zero_clock_stmt);
800 304 : if (table->meta_col_version_stmt) databasevm_finalize(table->meta_col_version_stmt);
801 304 : if (table->meta_site_id_stmt) databasevm_finalize(table->meta_site_id_stmt);
802 :
803 304 : if (table->real_col_values_stmt) databasevm_finalize(table->real_col_values_stmt);
804 304 : if (table->real_merge_delete_stmt) databasevm_finalize(table->real_merge_delete_stmt);
805 304 : if (table->real_merge_sentinel_stmt) databasevm_finalize(table->real_merge_sentinel_stmt);
806 :
807 304 : cloudsync_memory_free(table);
808 304 : }
809 :
810 304 : int table_add_stmts (cloudsync_table_context *table, int ncols) {
811 304 : int rc = DBRES_OK;
812 304 : char *sql = NULL;
813 304 : cloudsync_context *data = table->context;
814 :
815 : // META TABLE statements
816 :
817 : // CREATE TABLE IF NOT EXISTS \"%w_cloudsync\" (pk BLOB NOT NULL, col_name TEXT NOT NULL, col_version INTEGER, db_version INTEGER, site_id INTEGER DEFAULT 0, seq INTEGER, PRIMARY KEY (pk, col_name));
818 :
819 : // precompile the pk exists statement
820 : // we do not need an index on the pk column because it is already covered by the fact that it is part of the prikeys
821 : // EXPLAIN QUERY PLAN reports: SEARCH table_name USING PRIMARY KEY (pk=?)
822 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_ROW_EXISTS_BY_PK, table->meta_ref);
823 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
824 : DEBUG_SQL("meta_pkexists_stmt: %s", sql);
825 :
826 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_pkexists_stmt, DBFLAG_PERSISTENT);
827 304 : cloudsync_memory_free(sql);
828 304 : if (rc != DBRES_OK) goto cleanup;
829 :
830 : // precompile the update local sentinel statement
831 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPDATE_COL_BUMP_VERSION, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
832 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
833 : DEBUG_SQL("meta_sentinel_update_stmt: %s", sql);
834 :
835 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_sentinel_update_stmt, DBFLAG_PERSISTENT);
836 304 : cloudsync_memory_free(sql);
837 304 : if (rc != DBRES_OK) goto cleanup;
838 :
839 : // precompile the insert local sentinel statement
840 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPSERT_COL_INIT_OR_BUMP_VERSION, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE, table->meta_ref, table->meta_ref, table->meta_ref);
841 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
842 : DEBUG_SQL("meta_sentinel_insert_stmt: %s", sql);
843 :
844 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_sentinel_insert_stmt, DBFLAG_PERSISTENT);
845 304 : cloudsync_memory_free(sql);
846 304 : if (rc != DBRES_OK) goto cleanup;
847 :
848 : // precompile the insert/update local row statement
849 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION, table->meta_ref, table->meta_ref);
850 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
851 : DEBUG_SQL("meta_row_insert_update_stmt: %s", sql);
852 :
853 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_row_insert_update_stmt, DBFLAG_PERSISTENT);
854 304 : cloudsync_memory_free(sql);
855 304 : if (rc != DBRES_OK) goto cleanup;
856 :
857 : // precompile the delete rows from meta
858 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
859 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
860 : DEBUG_SQL("meta_row_drop_stmt: %s", sql);
861 :
862 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_row_drop_stmt, DBFLAG_PERSISTENT);
863 304 : cloudsync_memory_free(sql);
864 304 : if (rc != DBRES_OK) goto cleanup;
865 :
866 : // precompile the update rows from meta when pk changes
867 : // see https://github.com/sqliteai/sqlite-sync/blob/main/docs/PriKey.md for more details
868 304 : sql = sql_build_rekey_pk_and_reset_version_except_col(data, table->name, CLOUDSYNC_TOMBSTONE_VALUE);
869 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
870 : DEBUG_SQL("meta_update_move_stmt: %s", sql);
871 :
872 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_update_move_stmt, DBFLAG_PERSISTENT);
873 304 : cloudsync_memory_free(sql);
874 304 : if (rc != DBRES_OK) goto cleanup;
875 :
876 : // local cl
877 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_GET_COL_VERSION_OR_ROW_EXISTS, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE, table->meta_ref);
878 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
879 : DEBUG_SQL("meta_local_cl_stmt: %s", sql);
880 :
881 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_local_cl_stmt, DBFLAG_PERSISTENT);
882 304 : cloudsync_memory_free(sql);
883 304 : if (rc != DBRES_OK) goto cleanup;
884 :
885 : // rowid of the last inserted/updated row in the meta table
886 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_INSERT_RETURN_CHANGE_ID, table->meta_ref);
887 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
888 : DEBUG_SQL("meta_winner_clock_stmt: %s", sql);
889 :
890 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_winner_clock_stmt, DBFLAG_PERSISTENT);
891 304 : cloudsync_memory_free(sql);
892 304 : if (rc != DBRES_OK) goto cleanup;
893 :
894 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
895 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
896 : DEBUG_SQL("meta_merge_delete_drop: %s", sql);
897 :
898 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_merge_delete_drop, DBFLAG_PERSISTENT);
899 304 : cloudsync_memory_free(sql);
900 304 : if (rc != DBRES_OK) goto cleanup;
901 :
902 : // zero clock
903 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_TOMBSTONE_PK_EXCEPT_COL, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
904 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
905 : DEBUG_SQL("meta_zero_clock_stmt: %s", sql);
906 :
907 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_zero_clock_stmt, DBFLAG_PERSISTENT);
908 304 : cloudsync_memory_free(sql);
909 304 : if (rc != DBRES_OK) goto cleanup;
910 :
911 : // col_version
912 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_COL_VERSION_BY_PK_COL, table->meta_ref);
913 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
914 : DEBUG_SQL("meta_col_version_stmt: %s", sql);
915 :
916 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_col_version_stmt, DBFLAG_PERSISTENT);
917 304 : cloudsync_memory_free(sql);
918 304 : if (rc != DBRES_OK) goto cleanup;
919 :
920 : // site_id
921 304 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_SITE_ID_BY_PK_COL, table->meta_ref);
922 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
923 : DEBUG_SQL("meta_site_id_stmt: %s", sql);
924 :
925 304 : rc = databasevm_prepare(data, sql, (void **)&table->meta_site_id_stmt, DBFLAG_PERSISTENT);
926 304 : cloudsync_memory_free(sql);
927 304 : if (rc != DBRES_OK) goto cleanup;
928 :
929 : // REAL TABLE statements
930 :
931 : // precompile the get column value statement
932 304 : if (ncols > 0) {
933 265 : sql = sql_build_select_nonpk_by_pk(data, table->name, table->schema);
934 265 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
935 : DEBUG_SQL("real_col_values_stmt: %s", sql);
936 :
937 265 : rc = databasevm_prepare(data, sql, (void **)&table->real_col_values_stmt, DBFLAG_PERSISTENT);
938 265 : cloudsync_memory_free(sql);
939 265 : if (rc != DBRES_OK) goto cleanup;
940 265 : }
941 :
942 304 : sql = table_build_mergedelete_sql(table);
943 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
944 : DEBUG_SQL("real_merge_delete: %s", sql);
945 :
946 304 : rc = databasevm_prepare(data, sql, (void **)&table->real_merge_delete_stmt, DBFLAG_PERSISTENT);
947 304 : cloudsync_memory_free(sql);
948 304 : if (rc != DBRES_OK) goto cleanup;
949 :
950 304 : sql = table_build_mergeinsert_sql(table, NULL);
951 304 : if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
952 : DEBUG_SQL("real_merge_sentinel: %s", sql);
953 :
954 304 : rc = databasevm_prepare(data, sql, (void **)&table->real_merge_sentinel_stmt, DBFLAG_PERSISTENT);
955 304 : cloudsync_memory_free(sql);
956 304 : if (rc != DBRES_OK) goto cleanup;
957 :
958 : cleanup:
959 304 : if (rc != DBRES_OK) DEBUG_ALWAYS("table_add_stmts error: %d %s\n", rc, database_errmsg(data));
960 304 : return rc;
961 : }
962 :
963 206281 : cloudsync_table_context *table_lookup (cloudsync_context *data, const char *table_name) {
964 : DEBUG_DBFUNCTION("table_lookup %s", table_name);
965 :
966 206281 : if (table_name) {
967 207907 : for (int i=0; i<data->tables_count; ++i) {
968 207290 : if ((strcasecmp(data->tables[i]->name, table_name) == 0)) return data->tables[i];
969 1626 : }
970 617 : }
971 :
972 617 : return NULL;
973 206281 : }
974 :
975 188872 : void *table_column_lookup (cloudsync_table_context *table, const char *col_name, bool is_merge, int *index) {
976 : DEBUG_DBFUNCTION("table_column_lookup %s", col_name);
977 :
978 338218 : for (int i=0; i<table->ncols; ++i) {
979 338218 : if (strcasecmp(table->col_name[i], col_name) == 0) {
980 188872 : if (index) *index = i;
981 188872 : return (is_merge) ? table->col_merge_stmt[i] : table->col_value_stmt[i];
982 : }
983 149346 : }
984 :
985 0 : if (index) *index = -1;
986 0 : return NULL;
987 188872 : }
988 :
989 303 : int table_remove (cloudsync_context *data, cloudsync_table_context *table) {
990 303 : const char *table_name = table->name;
991 : DEBUG_DBFUNCTION("table_remove %s", table_name);
992 :
993 353 : for (int i = 0; i < data->tables_count; ++i) {
994 353 : cloudsync_table_context *t = data->tables[i];
995 :
996 : // pointer compare is fastest but fallback to strcasecmp if not same pointer
997 353 : if ((t == table) || ((strcasecmp(t->name, table_name) == 0))) {
998 303 : int last = data->tables_count - 1;
999 303 : data->tables[i] = data->tables[last]; // move last into the hole (keeps array dense)
1000 303 : data->tables[last] = NULL; // NULLify tail (as an extra security measure)
1001 303 : data->tables_count--;
1002 303 : return data->tables_count;
1003 : }
1004 50 : }
1005 :
1006 0 : return -1;
1007 303 : }
1008 :
1009 1119 : int table_add_to_context_cb (void *xdata, int ncols, char **values, char **names) {
1010 1119 : cloudsync_table_context *table = (cloudsync_table_context *)xdata;
1011 1119 : cloudsync_context *data = table->context;
1012 :
1013 1119 : int index = table->ncols;
1014 2237 : for (int i=0; i<ncols; i+=2) {
1015 1119 : const char *name = values[i];
1016 1119 : int cid = (int)strtol(values[i+1], NULL, 0);
1017 :
1018 1119 : table->col_id[index] = cid;
1019 1119 : table->col_name[index] = cloudsync_string_dup_lowercase(name);
1020 1119 : if (!table->col_name[index]) goto error;
1021 :
1022 1119 : char *sql = table_build_mergeinsert_sql(table, name);
1023 1119 : if (!sql) goto error;
1024 : DEBUG_SQL("col_merge_stmt[%d]: %s", index, sql);
1025 :
1026 1119 : int rc = databasevm_prepare(data, sql, (void **)&table->col_merge_stmt[index], DBFLAG_PERSISTENT);
1027 1119 : cloudsync_memory_free(sql);
1028 1119 : if (rc != DBRES_OK) goto error;
1029 1118 : if (!table->col_merge_stmt[index]) goto error;
1030 :
1031 1118 : sql = table_build_value_sql(table, name);
1032 1118 : if (!sql) goto error;
1033 : DEBUG_SQL("col_value_stmt[%d]: %s", index, sql);
1034 :
1035 1118 : rc = databasevm_prepare(data, sql, (void **)&table->col_value_stmt[index], DBFLAG_PERSISTENT);
1036 1118 : cloudsync_memory_free(sql);
1037 1118 : if (rc != DBRES_OK) goto error;
1038 1118 : if (!table->col_value_stmt[index]) goto error;
1039 1118 : }
1040 1118 : table->ncols += 1;
1041 :
1042 1118 : return 0;
1043 :
1044 : error:
1045 : // clean up partially-initialized entry at index
1046 1 : if (table->col_name[index]) {cloudsync_memory_free(table->col_name[index]); table->col_name[index] = NULL;}
1047 1 : if (table->col_merge_stmt[index]) {databasevm_finalize(table->col_merge_stmt[index]); table->col_merge_stmt[index] = NULL;}
1048 1 : if (table->col_value_stmt[index]) {databasevm_finalize(table->col_value_stmt[index]); table->col_value_stmt[index] = NULL;}
1049 1 : return 1;
1050 1119 : }
1051 :
1052 304 : bool table_ensure_capacity (cloudsync_context *data) {
1053 304 : if (data->tables_count < data->tables_cap) return true;
1054 :
1055 0 : int new_cap = data->tables_cap ? data->tables_cap * 2 : CLOUDSYNC_INIT_NTABLES;
1056 0 : size_t bytes = (size_t)new_cap * sizeof(*data->tables);
1057 0 : void *p = cloudsync_memory_realloc(data->tables, bytes);
1058 0 : if (!p) return false;
1059 :
1060 0 : data->tables = (cloudsync_table_context **)p;
1061 0 : data->tables_cap = new_cap;
1062 0 : return true;
1063 304 : }
1064 :
1065 309 : bool table_add_to_context (cloudsync_context *data, table_algo algo, const char *table_name) {
1066 : DEBUG_DBFUNCTION("cloudsync_context_add_table %s", table_name);
1067 :
1068 : // Check if table already initialized in this connection's context.
1069 : // Note: This prevents same-connection duplicate initialization.
1070 : // SQLite clients cannot distinguish schemas, so having 'public.users'
1071 : // and 'auth.users' would cause sync ambiguity. Users should avoid
1072 : // initializing tables with the same name in different schemas.
1073 : // If two concurrent connections initialize tables with the same name
1074 : // in different schemas, the behavior is undefined.
1075 309 : cloudsync_table_context *table = table_lookup(data, table_name);
1076 309 : if (table) return true;
1077 :
1078 : // check for space availability
1079 304 : if (!table_ensure_capacity(data)) return false;
1080 :
1081 : // setup a new table
1082 304 : table = table_create(data, table_name, algo);
1083 304 : if (!table) return false;
1084 :
1085 : // fill remaining metadata in the table
1086 304 : int count = database_count_pk(data, table_name, false, table->schema);
1087 304 : if (count < 0) {cloudsync_set_dberror(data); goto abort_add_table;}
1088 304 : table->npks = count;
1089 304 : if (table->npks == 0) {
1090 : #if CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
1091 0 : goto abort_add_table;
1092 : #else
1093 : table->rowid_only = true;
1094 : table->npks = 1; // rowid
1095 : #endif
1096 : }
1097 :
1098 304 : int ncols = database_count_nonpk(data, table_name, table->schema);
1099 304 : if (ncols < 0) {cloudsync_set_dberror(data); goto abort_add_table;}
1100 304 : int rc = table_add_stmts(table, ncols);
1101 304 : if (rc != DBRES_OK) goto abort_add_table;
1102 :
1103 : // a table with only pk(s) is totally legal
1104 304 : if (ncols > 0) {
1105 265 : table->col_name = (char **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(char *) * ncols));
1106 265 : if (!table->col_name) goto abort_add_table;
1107 :
1108 265 : table->col_id = (int *)cloudsync_memory_zeroalloc((uint64_t)(sizeof(int) * ncols));
1109 265 : if (!table->col_id) goto abort_add_table;
1110 :
1111 265 : table->col_merge_stmt = (dbvm_t **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(void *) * ncols));
1112 265 : if (!table->col_merge_stmt) goto abort_add_table;
1113 :
1114 265 : table->col_value_stmt = (dbvm_t **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(void *) * ncols));
1115 265 : if (!table->col_value_stmt) goto abort_add_table;
1116 :
1117 265 : table->col_algo = (col_algo_t *)cloudsync_memory_zeroalloc((uint64_t)(sizeof(col_algo_t) * ncols));
1118 265 : if (!table->col_algo) goto abort_add_table;
1119 :
1120 265 : table->col_delimiter = (char **)cloudsync_memory_zeroalloc((uint64_t)(sizeof(char *) * ncols));
1121 265 : if (!table->col_delimiter) goto abort_add_table;
1122 :
1123 : // Pass empty string when schema is NULL; SQL will fall back to current_schema()
1124 265 : const char *schema = table->schema ? table->schema : "";
1125 530 : char *sql = cloudsync_memory_mprintf(SQL_PRAGMA_TABLEINFO_LIST_NONPK_NAME_CID,
1126 265 : table_name, schema, table_name, schema);
1127 265 : if (!sql) goto abort_add_table;
1128 265 : rc = database_exec_callback(data, sql, table_add_to_context_cb, (void *)table);
1129 265 : cloudsync_memory_free(sql);
1130 265 : if (rc == DBRES_ABORT) goto abort_add_table;
1131 264 : }
1132 :
1133 : // append newly created table
1134 303 : data->tables[data->tables_count++] = table;
1135 303 : return true;
1136 :
1137 : abort_add_table:
1138 1 : table_free(table);
1139 1 : return false;
1140 309 : }
1141 :
1142 0 : dbvm_t *cloudsync_colvalue_stmt (cloudsync_context *data, const char *tbl_name, bool *persistent) {
1143 0 : dbvm_t *vm = NULL;
1144 0 : *persistent = false;
1145 :
1146 0 : cloudsync_table_context *table = table_lookup(data, tbl_name);
1147 0 : if (table) {
1148 0 : char *col_name = NULL;
1149 0 : if (table->ncols > 0) {
1150 0 : col_name = table->col_name[0];
1151 : // retrieve col_value precompiled statement
1152 0 : vm = table_column_lookup(table, col_name, false, NULL);
1153 0 : *persistent = true;
1154 0 : } else {
1155 0 : char *sql = table_build_value_sql(table, "*");
1156 0 : databasevm_prepare(data, sql, (void **)&vm, 0);
1157 0 : cloudsync_memory_free(sql);
1158 0 : *persistent = false;
1159 : }
1160 0 : }
1161 :
1162 0 : return vm;
1163 : }
1164 :
1165 8148 : bool table_enabled (cloudsync_table_context *table) {
1166 8148 : return table->enabled;
1167 : }
1168 :
1169 6 : void table_set_enabled (cloudsync_table_context *table, bool value) {
1170 6 : table->enabled = value;
1171 6 : }
1172 :
1173 23933 : int table_count_cols (cloudsync_table_context *table) {
1174 23933 : return table->ncols;
1175 : }
1176 :
1177 8049 : int table_count_pks (cloudsync_table_context *table) {
1178 8049 : return table->npks;
1179 : }
1180 :
1181 36058 : const char *table_colname (cloudsync_table_context *table, int index) {
1182 36058 : return table->col_name[index];
1183 : }
1184 :
1185 4856 : bool table_pk_exists (cloudsync_table_context *table, const char *value, size_t len) {
1186 : // check if a row with the same primary key already exists
1187 : // if so, this means the row might have been previously deleted (sentinel)
1188 4856 : return (dbvm_count(table->meta_pkexists_stmt, value, len, DBTYPE_BLOB) > 0);
1189 : }
1190 :
1191 0 : char **table_pknames (cloudsync_table_context *table) {
1192 0 : return table->pk_name;
1193 : }
1194 :
1195 23 : void table_set_pknames (cloudsync_table_context *table, char **pknames) {
1196 23 : table_pknames_free(table->pk_name, table->npks);
1197 23 : table->pk_name = pknames;
1198 23 : }
1199 :
1200 51495 : bool table_algo_isgos (cloudsync_table_context *table) {
1201 51495 : return (table->algo == table_algo_crdt_gos);
1202 : }
1203 :
1204 0 : const char *table_schema (cloudsync_table_context *table) {
1205 0 : return table->schema;
1206 : }
1207 :
1208 : // MARK: - Merge Insert -
1209 :
1210 48345 : int64_t merge_get_local_cl (cloudsync_table_context *table, const char *pk, int pklen) {
1211 48345 : dbvm_t *vm = table->meta_local_cl_stmt;
1212 48345 : int64_t result = -1;
1213 :
1214 48345 : int rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1215 48345 : if (rc != DBRES_OK) goto cleanup;
1216 :
1217 48345 : rc = databasevm_bind_blob(vm, 2, (const void *)pk, pklen);
1218 48345 : if (rc != DBRES_OK) goto cleanup;
1219 :
1220 48345 : rc = databasevm_step(vm);
1221 48345 : if (rc == DBRES_ROW) result = database_column_int(vm, 0);
1222 0 : else if (rc == DBRES_DONE) result = 0;
1223 :
1224 : cleanup:
1225 48345 : if (result == -1) cloudsync_set_dberror(table->context);
1226 48345 : dbvm_reset(vm);
1227 48345 : return result;
1228 : }
1229 :
1230 47333 : int merge_get_col_version (cloudsync_table_context *table, const char *col_name, const char *pk, int pklen, int64_t *version) {
1231 47333 : dbvm_t *vm = table->meta_col_version_stmt;
1232 :
1233 47333 : int rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1234 47333 : if (rc != DBRES_OK) goto cleanup;
1235 :
1236 47333 : rc = databasevm_bind_text(vm, 2, col_name, -1);
1237 47333 : if (rc != DBRES_OK) goto cleanup;
1238 :
1239 47333 : rc = databasevm_step(vm);
1240 72469 : if (rc == DBRES_ROW) {
1241 25136 : *version = database_column_int(vm, 0);
1242 25136 : rc = DBRES_OK;
1243 25136 : }
1244 :
1245 : cleanup:
1246 47333 : if ((rc != DBRES_OK) && (rc != DBRES_DONE)) cloudsync_set_dberror(table->context);
1247 47333 : dbvm_reset(vm);
1248 47333 : return rc;
1249 : }
1250 :
1251 27235 : int merge_set_winner_clock (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pk_len, const char *colname, int64_t col_version, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1252 :
1253 : // get/set site_id
1254 27235 : dbvm_t *vm = data->getset_siteid_stmt;
1255 27235 : int rc = databasevm_bind_blob(vm, 1, (const void *)site_id, site_len);
1256 27235 : if (rc != DBRES_OK) goto cleanup_merge;
1257 :
1258 27235 : rc = databasevm_step(vm);
1259 27235 : if (rc != DBRES_ROW) goto cleanup_merge;
1260 :
1261 27235 : int64_t ord = database_column_int(vm, 0);
1262 27235 : dbvm_reset(vm);
1263 :
1264 27235 : vm = table->meta_winner_clock_stmt;
1265 27235 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pk_len);
1266 27235 : if (rc != DBRES_OK) goto cleanup_merge;
1267 :
1268 27235 : rc = databasevm_bind_text(vm, 2, (colname) ? colname : CLOUDSYNC_TOMBSTONE_VALUE, -1);
1269 27235 : if (rc != DBRES_OK) goto cleanup_merge;
1270 :
1271 27235 : rc = databasevm_bind_int(vm, 3, col_version);
1272 27235 : if (rc != DBRES_OK) goto cleanup_merge;
1273 :
1274 27235 : rc = databasevm_bind_int(vm, 4, db_version);
1275 27235 : if (rc != DBRES_OK) goto cleanup_merge;
1276 :
1277 27235 : rc = databasevm_bind_int(vm, 5, seq);
1278 27235 : if (rc != DBRES_OK) goto cleanup_merge;
1279 :
1280 27235 : rc = databasevm_bind_int(vm, 6, ord);
1281 27235 : if (rc != DBRES_OK) goto cleanup_merge;
1282 :
1283 27235 : rc = databasevm_step(vm);
1284 54470 : if (rc == DBRES_ROW) {
1285 27235 : *rowid = database_column_int(vm, 0);
1286 27235 : rc = DBRES_OK;
1287 27235 : }
1288 :
1289 : cleanup_merge:
1290 27235 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1291 27235 : dbvm_reset(vm);
1292 27235 : return rc;
1293 : }
1294 :
1295 : // MARK: - Deferred column-batch merge functions -
1296 :
1297 22939 : static int merge_pending_add (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, const char *col_name, dbvalue_t *col_value, int64_t col_version, int64_t db_version, const char *site_id, int site_len, int64_t seq) {
1298 22939 : merge_pending_batch *batch = data->pending_batch;
1299 :
1300 : // Store table and PK on first entry
1301 22939 : if (batch->table == NULL) {
1302 8760 : batch->table = table;
1303 8760 : batch->pk = (char *)cloudsync_memory_alloc(pklen);
1304 8760 : if (!batch->pk) return cloudsync_set_error(data, "merge_pending_add: out of memory for pk", DBRES_NOMEM);
1305 8760 : memcpy(batch->pk, pk, pklen);
1306 8760 : batch->pk_len = pklen;
1307 8760 : }
1308 :
1309 : // Ensure capacity
1310 22939 : if (batch->count >= batch->capacity) {
1311 537 : int new_cap = batch->capacity ? batch->capacity * 2 : 8;
1312 537 : merge_pending_entry *new_entries = (merge_pending_entry *)cloudsync_memory_realloc(batch->entries, new_cap * sizeof(merge_pending_entry));
1313 537 : if (!new_entries) return cloudsync_set_error(data, "merge_pending_add: out of memory for entries", DBRES_NOMEM);
1314 537 : batch->entries = new_entries;
1315 537 : batch->capacity = new_cap;
1316 537 : }
1317 :
1318 : // Resolve col_name to a stable pointer from the table context
1319 : // (the incoming col_name may point to VM-owned memory that gets freed on reset)
1320 22939 : int col_idx = -1;
1321 22939 : table_column_lookup(table, col_name, true, &col_idx);
1322 22939 : const char *stable_col_name = (col_idx >= 0) ? table_colname(table, col_idx) : NULL;
1323 22939 : if (!stable_col_name) return cloudsync_set_error(data, "merge_pending_add: column not found in table context", DBRES_ERROR);
1324 :
1325 22939 : merge_pending_entry *e = &batch->entries[batch->count];
1326 22939 : e->col_name = stable_col_name;
1327 22939 : e->col_value = col_value ? (dbvalue_t *)database_value_dup(col_value) : NULL;
1328 22939 : e->col_version = col_version;
1329 22939 : e->db_version = db_version;
1330 22939 : e->site_id_len = (site_len <= (int)sizeof(e->site_id)) ? site_len : (int)sizeof(e->site_id);
1331 22939 : memcpy(e->site_id, site_id, e->site_id_len);
1332 22939 : e->seq = seq;
1333 :
1334 22939 : batch->count++;
1335 22939 : return DBRES_OK;
1336 22939 : }
1337 :
1338 20684 : static void merge_pending_free_entries (merge_pending_batch *batch) {
1339 20684 : if (batch->entries) {
1340 38265 : for (int i = 0; i < batch->count; i++) {
1341 22939 : if (batch->entries[i].col_value) {
1342 22939 : database_value_free(batch->entries[i].col_value);
1343 22939 : batch->entries[i].col_value = NULL;
1344 22939 : }
1345 22939 : }
1346 15326 : }
1347 20684 : if (batch->pk) {
1348 8792 : cloudsync_memory_free(batch->pk);
1349 8792 : batch->pk = NULL;
1350 8792 : }
1351 20684 : batch->table = NULL;
1352 20684 : batch->pk_len = 0;
1353 20684 : batch->cl = 0;
1354 20684 : batch->sentinel_pending = false;
1355 20684 : batch->row_exists = false;
1356 20684 : batch->count = 0;
1357 20684 : }
1358 :
1359 20673 : static int merge_flush_pending (cloudsync_context *data) {
1360 20673 : merge_pending_batch *batch = data->pending_batch;
1361 20673 : if (!batch) return DBRES_OK;
1362 :
1363 20673 : int rc = DBRES_OK;
1364 20673 : bool flush_savepoint = false;
1365 :
1366 : // Nothing to write — handle sentinel-only case or skip
1367 20673 : if (batch->count == 0 && !(batch->sentinel_pending && batch->table)) {
1368 11881 : goto cleanup;
1369 : }
1370 :
1371 : // Wrap database operations in a savepoint so that on failure (e.g. RLS
1372 : // denial) the rollback properly releases all executor resources (open
1373 : // relations, snapshots, plan cache) acquired during the failed statement.
1374 8792 : flush_savepoint = (database_begin_savepoint(data, "merge_flush") == DBRES_OK);
1375 :
1376 8792 : if (batch->count == 0) {
1377 : // Sentinel with no winning columns (PK-only row)
1378 30 : dbvm_t *vm = batch->table->real_merge_sentinel_stmt;
1379 30 : rc = pk_decode_prikey(batch->pk, (size_t)batch->pk_len, pk_decode_bind_callback, vm);
1380 30 : if (rc < 0) {
1381 0 : cloudsync_set_dberror(data);
1382 0 : dbvm_reset(vm);
1383 0 : goto cleanup;
1384 : }
1385 30 : SYNCBIT_SET(data);
1386 30 : rc = databasevm_step(vm);
1387 30 : dbvm_reset(vm);
1388 30 : SYNCBIT_RESET(data);
1389 30 : if (rc == DBRES_DONE) rc = DBRES_OK;
1390 30 : if (rc != DBRES_OK) {
1391 0 : cloudsync_set_dberror(data);
1392 0 : goto cleanup;
1393 : }
1394 30 : goto cleanup;
1395 : }
1396 :
1397 : // Check if cached prepared statement can be reused
1398 8762 : cloudsync_table_context *table = batch->table;
1399 8762 : dbvm_t *vm = NULL;
1400 8762 : bool cache_hit = false;
1401 :
1402 16680 : if (batch->cached_vm &&
1403 8225 : batch->cached_row_exists == batch->row_exists &&
1404 7918 : batch->cached_col_count == batch->count) {
1405 7889 : cache_hit = true;
1406 29116 : for (int i = 0; i < batch->count; i++) {
1407 21253 : if (batch->cached_col_names[i] != batch->entries[i].col_name) {
1408 26 : cache_hit = false;
1409 26 : break;
1410 : }
1411 21227 : }
1412 7889 : }
1413 :
1414 8762 : if (cache_hit) {
1415 7863 : vm = batch->cached_vm;
1416 7863 : dbvm_reset(vm);
1417 7863 : } else {
1418 : // Invalidate old cache
1419 899 : if (batch->cached_vm) {
1420 362 : databasevm_finalize(batch->cached_vm);
1421 362 : batch->cached_vm = NULL;
1422 362 : }
1423 :
1424 : // Build multi-column SQL
1425 899 : const char **colnames = (const char **)cloudsync_memory_alloc(batch->count * sizeof(const char *));
1426 899 : if (!colnames) {
1427 0 : rc = cloudsync_set_error(data, "merge_flush_pending: out of memory", DBRES_NOMEM);
1428 0 : goto cleanup;
1429 : }
1430 2611 : for (int i = 0; i < batch->count; i++) {
1431 1712 : colnames[i] = batch->entries[i].col_name;
1432 1712 : }
1433 :
1434 899 : char *sql = batch->row_exists
1435 440 : ? sql_build_update_pk_and_multi_cols(data, table->name, colnames, batch->count, table->schema)
1436 459 : : sql_build_upsert_pk_and_multi_cols(data, table->name, colnames, batch->count, table->schema);
1437 899 : cloudsync_memory_free(colnames);
1438 :
1439 899 : if (!sql) {
1440 0 : rc = cloudsync_set_error(data, "merge_flush_pending: unable to build multi-column upsert SQL", DBRES_ERROR);
1441 0 : goto cleanup;
1442 : }
1443 :
1444 899 : rc = databasevm_prepare(data, sql, &vm, 0);
1445 899 : cloudsync_memory_free(sql);
1446 899 : if (rc != DBRES_OK) {
1447 0 : rc = cloudsync_set_error(data, "merge_flush_pending: unable to prepare statement", rc);
1448 0 : goto cleanup;
1449 : }
1450 :
1451 : // Update cache
1452 899 : batch->cached_vm = vm;
1453 899 : batch->cached_row_exists = batch->row_exists;
1454 899 : batch->cached_col_count = batch->count;
1455 : // Reallocate cached_col_names if needed
1456 899 : if (batch->cached_col_count > 0) {
1457 899 : const char **new_names = (const char **)cloudsync_memory_realloc(
1458 899 : batch->cached_col_names, batch->count * sizeof(const char *));
1459 899 : if (new_names) {
1460 2611 : for (int i = 0; i < batch->count; i++) {
1461 1712 : new_names[i] = batch->entries[i].col_name;
1462 1712 : }
1463 899 : batch->cached_col_names = new_names;
1464 899 : }
1465 899 : }
1466 : }
1467 :
1468 : // Bind PKs (positions 1..npks)
1469 8762 : int npks = pk_decode_prikey(batch->pk, (size_t)batch->pk_len, pk_decode_bind_callback, vm);
1470 8762 : if (npks < 0) {
1471 0 : cloudsync_set_dberror(data);
1472 0 : dbvm_reset(vm);
1473 0 : rc = DBRES_ERROR;
1474 0 : goto cleanup;
1475 : }
1476 :
1477 : // Bind column values (positions npks+1..npks+count)
1478 31701 : for (int i = 0; i < batch->count; i++) {
1479 22939 : merge_pending_entry *e = &batch->entries[i];
1480 22939 : int bind_idx = npks + 1 + i;
1481 22939 : if (e->col_value) {
1482 22939 : rc = databasevm_bind_value(vm, bind_idx, e->col_value);
1483 22939 : } else {
1484 0 : rc = databasevm_bind_null(vm, bind_idx);
1485 : }
1486 22939 : if (rc != DBRES_OK) {
1487 0 : cloudsync_set_dberror(data);
1488 0 : dbvm_reset(vm);
1489 0 : goto cleanup;
1490 : }
1491 22939 : }
1492 :
1493 : // Execute with SYNCBIT and GOS handling
1494 8762 : if (table->algo == table_algo_crdt_gos) table->enabled = 0;
1495 8762 : SYNCBIT_SET(data);
1496 8762 : rc = databasevm_step(vm);
1497 8762 : dbvm_reset(vm);
1498 8762 : SYNCBIT_RESET(data);
1499 8762 : if (table->algo == table_algo_crdt_gos) table->enabled = 1;
1500 :
1501 8762 : if (rc != DBRES_DONE) {
1502 3 : cloudsync_set_dberror(data);
1503 3 : goto cleanup;
1504 : }
1505 8759 : rc = DBRES_OK;
1506 :
1507 : // Call merge_set_winner_clock for each buffered entry
1508 8759 : int64_t rowid = 0;
1509 31690 : for (int i = 0; i < batch->count; i++) {
1510 22931 : merge_pending_entry *e = &batch->entries[i];
1511 45862 : int clock_rc = merge_set_winner_clock(data, table, batch->pk, batch->pk_len,
1512 22931 : e->col_name, e->col_version, e->db_version,
1513 22931 : (const char *)e->site_id, e->site_id_len,
1514 22931 : e->seq, &rowid);
1515 22931 : if (clock_rc != DBRES_OK) {
1516 0 : rc = clock_rc;
1517 0 : goto cleanup;
1518 : }
1519 31690 : }
1520 :
1521 : cleanup:
1522 20673 : merge_pending_free_entries(batch);
1523 20673 : if (flush_savepoint) {
1524 8792 : if (rc == DBRES_OK) database_commit_savepoint(data, "merge_flush");
1525 3 : else database_rollback_savepoint(data, "merge_flush");
1526 8792 : }
1527 20673 : return rc;
1528 20673 : }
1529 :
1530 3667 : int merge_insert_col (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *col_name, dbvalue_t *col_value, int64_t col_version, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1531 : int index;
1532 3667 : dbvm_t *vm = table_column_lookup(table, col_name, true, &index);
1533 3667 : if (vm == NULL) return cloudsync_set_error(data, "Unable to retrieve column merge precompiled statement in merge_insert_col", DBRES_MISUSE);
1534 :
1535 : // INSERT INTO table (pk1, pk2, col_name) VALUES (?, ?, ?) ON CONFLICT DO UPDATE SET col_name=?;"
1536 :
1537 : // bind primary key(s)
1538 3667 : int rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, vm);
1539 3667 : if (rc < 0) {
1540 0 : cloudsync_set_dberror(data);
1541 0 : dbvm_reset(vm);
1542 0 : return rc;
1543 : }
1544 :
1545 : // bind value (always bind all expected parameters for correct prepared statement handling)
1546 3667 : if (col_value) {
1547 3667 : rc = databasevm_bind_value(vm, table->npks+1, col_value);
1548 3667 : if (rc == DBRES_OK) rc = databasevm_bind_value(vm, table->npks+2, col_value);
1549 3667 : } else {
1550 0 : rc = databasevm_bind_null(vm, table->npks+1);
1551 0 : if (rc == DBRES_OK) rc = databasevm_bind_null(vm, table->npks+2);
1552 : }
1553 3667 : if (rc != DBRES_OK) {
1554 0 : cloudsync_set_dberror(data);
1555 0 : dbvm_reset(vm);
1556 0 : return rc;
1557 : }
1558 :
1559 : // perform real operation and disable triggers
1560 :
1561 : // in case of GOS we reused the table->col_merge_stmt statement
1562 : // which looks like: INSERT INTO table (pk1, pk2, col_name) VALUES (?, ?, ?) ON CONFLICT DO UPDATE SET col_name=?;"
1563 : // but the UPDATE in the CONFLICT statement would return SQLITE_CONSTRAINT because the trigger raises the error
1564 : // the trick is to disable that trigger before executing the statement
1565 3667 : if (table->algo == table_algo_crdt_gos) table->enabled = 0;
1566 3667 : SYNCBIT_SET(data);
1567 3667 : rc = databasevm_step(vm);
1568 : DEBUG_MERGE("merge_insert(%02x%02x): %s (%d)", data->site_id[UUID_LEN-2], data->site_id[UUID_LEN-1], databasevm_sql(vm), rc);
1569 3667 : dbvm_reset(vm);
1570 3667 : SYNCBIT_RESET(data);
1571 3667 : if (table->algo == table_algo_crdt_gos) table->enabled = 1;
1572 :
1573 3667 : if (rc != DBRES_DONE) {
1574 0 : cloudsync_set_dberror(data);
1575 0 : return rc;
1576 : }
1577 :
1578 3667 : return merge_set_winner_clock(data, table, pk, pklen, col_name, col_version, db_version, site_id, site_len, seq, rowid);
1579 3667 : }
1580 :
1581 167 : int merge_delete (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, const char *colname, int64_t cl, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1582 167 : int rc = DBRES_OK;
1583 :
1584 : // reset return value
1585 167 : *rowid = 0;
1586 :
1587 : // bind pk
1588 167 : dbvm_t *vm = table->real_merge_delete_stmt;
1589 167 : rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, vm);
1590 167 : if (rc < 0) {
1591 0 : rc = cloudsync_set_dberror(data);
1592 0 : dbvm_reset(vm);
1593 0 : return rc;
1594 : }
1595 :
1596 : // perform real operation and disable triggers
1597 167 : SYNCBIT_SET(data);
1598 167 : rc = databasevm_step(vm);
1599 : DEBUG_MERGE("merge_delete(%02x%02x): %s (%d)", data->site_id[UUID_LEN-2], data->site_id[UUID_LEN-1], databasevm_sql(vm), rc);
1600 167 : dbvm_reset(vm);
1601 167 : SYNCBIT_RESET(data);
1602 167 : if (rc == DBRES_DONE) rc = DBRES_OK;
1603 167 : if (rc != DBRES_OK) {
1604 0 : cloudsync_set_dberror(data);
1605 0 : return rc;
1606 : }
1607 :
1608 167 : rc = merge_set_winner_clock(data, table, pk, pklen, colname, cl, db_version, site_id, site_len, seq, rowid);
1609 167 : if (rc != DBRES_OK) return rc;
1610 :
1611 : // drop clocks _after_ setting the winner clock so we don't lose track of the max db_version!!
1612 : // this must never come before `set_winner_clock`
1613 167 : vm = table->meta_merge_delete_drop;
1614 167 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1615 167 : if (rc == DBRES_OK) rc = databasevm_step(vm);
1616 167 : dbvm_reset(vm);
1617 :
1618 167 : if (rc == DBRES_DONE) rc = DBRES_OK;
1619 167 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1620 167 : return rc;
1621 167 : }
1622 :
1623 58 : int merge_zeroclock_on_resurrect(cloudsync_table_context *table, int64_t db_version, const char *pk, int pklen) {
1624 58 : dbvm_t *vm = table->meta_zero_clock_stmt;
1625 :
1626 58 : int rc = databasevm_bind_int(vm, 1, db_version);
1627 58 : if (rc != DBRES_OK) goto cleanup;
1628 :
1629 58 : rc = databasevm_bind_blob(vm, 2, (const void *)pk, pklen);
1630 58 : if (rc != DBRES_OK) goto cleanup;
1631 :
1632 58 : rc = databasevm_step(vm);
1633 58 : if (rc == DBRES_DONE) rc = DBRES_OK;
1634 :
1635 : cleanup:
1636 58 : if (rc != DBRES_OK) cloudsync_set_dberror(table->context);
1637 58 : dbvm_reset(vm);
1638 58 : return rc;
1639 : }
1640 :
1641 : // executed only if insert_cl == local_cl
1642 47333 : int merge_did_cid_win (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, dbvalue_t *insert_value, const char *site_id, int site_len, const char *col_name, int64_t col_version, bool *didwin_flag) {
1643 :
1644 47333 : if (col_name == NULL) col_name = CLOUDSYNC_TOMBSTONE_VALUE;
1645 :
1646 : int64_t local_version;
1647 47333 : int rc = merge_get_col_version(table, col_name, pk, pklen, &local_version);
1648 47333 : if (rc == DBRES_DONE) {
1649 : // no rows returned, the incoming change wins if there's nothing there locally
1650 22197 : *didwin_flag = true;
1651 22197 : return DBRES_OK;
1652 : }
1653 25136 : if (rc != DBRES_OK) return rc;
1654 :
1655 : // rc == DBRES_OK, means that a row with a version exists
1656 25136 : if (local_version != col_version) {
1657 2005 : if (col_version > local_version) {*didwin_flag = true; return DBRES_OK;}
1658 746 : if (col_version < local_version) {*didwin_flag = false; return DBRES_OK;}
1659 0 : }
1660 :
1661 : // rc == DBRES_ROW and col_version == local_version, need to compare values
1662 :
1663 : // retrieve col_value precompiled statement
1664 23131 : bool is_block_col = block_is_block_colname(col_name) && table_has_block_cols(table);
1665 : dbvm_t *vm;
1666 23131 : if (is_block_col) {
1667 : // Block column: read value from blocks table (pk + col_name bindings)
1668 43 : vm = table_block_value_read_stmt(table);
1669 43 : if (!vm) return cloudsync_set_error(data, "Unable to retrieve block value read statement in merge_did_cid_win", DBRES_ERROR);
1670 43 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1671 43 : if (rc != DBRES_OK) { dbvm_reset(vm); return cloudsync_set_dberror(data); }
1672 43 : rc = databasevm_bind_text(vm, 2, col_name, -1);
1673 43 : if (rc != DBRES_OK) { dbvm_reset(vm); return cloudsync_set_dberror(data); }
1674 43 : } else {
1675 23088 : vm = table_column_lookup(table, col_name, false, NULL);
1676 23088 : if (!vm) return cloudsync_set_error(data, "Unable to retrieve column value precompiled statement in merge_did_cid_win", DBRES_ERROR);
1677 :
1678 : // bind primary key values
1679 23088 : rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, (void *)vm);
1680 23088 : if (rc < 0) {
1681 0 : rc = cloudsync_set_dberror(data);
1682 0 : dbvm_reset(vm);
1683 0 : return rc;
1684 : }
1685 : }
1686 :
1687 : // execute vm
1688 : dbvalue_t *local_value;
1689 23131 : rc = databasevm_step(vm);
1690 23131 : if (rc == DBRES_DONE) {
1691 : // meta entry exists but the actual value is missing
1692 : // we should allow the value_compare function to make a decision
1693 : // value_compare has been modified to handle the case where lvalue is NULL
1694 2 : local_value = NULL;
1695 2 : rc = DBRES_OK;
1696 23131 : } else if (rc == DBRES_ROW) {
1697 23129 : local_value = database_column_value(vm, 0);
1698 23129 : rc = DBRES_OK;
1699 23129 : } else {
1700 0 : goto cleanup;
1701 : }
1702 :
1703 : // compare values
1704 23131 : int ret = dbutils_value_compare(insert_value, local_value);
1705 : // reset after compare, otherwise local value would be deallocated
1706 23131 : dbvm_reset(vm);
1707 23131 : vm = NULL;
1708 :
1709 23131 : bool compare_site_id = (ret == 0 && data->merge_equal_values == true);
1710 23131 : if (!compare_site_id) {
1711 23131 : *didwin_flag = (ret > 0);
1712 23131 : goto cleanup;
1713 : }
1714 :
1715 : // values are the same and merge_equal_values is true
1716 0 : vm = table->meta_site_id_stmt;
1717 0 : rc = databasevm_bind_blob(vm, 1, (const void *)pk, pklen);
1718 0 : if (rc != DBRES_OK) goto cleanup;
1719 :
1720 0 : rc = databasevm_bind_text(vm, 2, col_name, -1);
1721 0 : if (rc != DBRES_OK) goto cleanup;
1722 :
1723 0 : rc = databasevm_step(vm);
1724 0 : if (rc == DBRES_ROW) {
1725 0 : const void *local_site_id = database_column_blob(vm, 0, NULL);
1726 0 : if (!local_site_id) {
1727 0 : dbvm_reset(vm);
1728 0 : return cloudsync_set_error(data, "NULL site_id in cloudsync table, table is probably corrupted", DBRES_ERROR);
1729 : }
1730 0 : ret = memcmp(site_id, local_site_id, site_len);
1731 0 : *didwin_flag = (ret > 0);
1732 0 : dbvm_reset(vm);
1733 0 : return DBRES_OK;
1734 : }
1735 :
1736 : // handle error condition here
1737 0 : dbvm_reset(vm);
1738 0 : return cloudsync_set_error(data, "Unable to find site_id for previous change, cloudsync table is probably corrupted", DBRES_ERROR);
1739 :
1740 : cleanup:
1741 23131 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1742 23131 : dbvm_reset(vm);
1743 23131 : return rc;
1744 47333 : }
1745 :
1746 58 : int merge_sentinel_only_insert (cloudsync_context *data, cloudsync_table_context *table, const char *pk, int pklen, int64_t cl, int64_t db_version, const char *site_id, int site_len, int64_t seq, int64_t *rowid) {
1747 :
1748 : // reset return value
1749 58 : *rowid = 0;
1750 :
1751 58 : if (data->pending_batch == NULL) {
1752 : // Immediate mode: execute base table INSERT
1753 0 : dbvm_t *vm = table->real_merge_sentinel_stmt;
1754 0 : int rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, vm);
1755 0 : if (rc < 0) {
1756 0 : rc = cloudsync_set_dberror(data);
1757 0 : dbvm_reset(vm);
1758 0 : return rc;
1759 : }
1760 :
1761 0 : SYNCBIT_SET(data);
1762 0 : rc = databasevm_step(vm);
1763 0 : dbvm_reset(vm);
1764 0 : SYNCBIT_RESET(data);
1765 0 : if (rc == DBRES_DONE) rc = DBRES_OK;
1766 0 : if (rc != DBRES_OK) {
1767 0 : cloudsync_set_dberror(data);
1768 0 : return rc;
1769 : }
1770 0 : } else {
1771 : // Batch mode: skip base table INSERT, the batch flush will create the row
1772 58 : merge_pending_batch *batch = data->pending_batch;
1773 58 : batch->sentinel_pending = true;
1774 58 : if (batch->table == NULL) {
1775 32 : batch->table = table;
1776 32 : batch->pk = (char *)cloudsync_memory_alloc(pklen);
1777 32 : if (!batch->pk) return cloudsync_set_error(data, "merge_sentinel_only_insert: out of memory for pk", DBRES_NOMEM);
1778 32 : memcpy(batch->pk, pk, pklen);
1779 32 : batch->pk_len = pklen;
1780 32 : }
1781 : }
1782 :
1783 : // Metadata operations always execute regardless of batch mode
1784 58 : int rc = merge_zeroclock_on_resurrect(table, db_version, pk, pklen);
1785 58 : if (rc != DBRES_OK) return rc;
1786 :
1787 58 : return merge_set_winner_clock(data, table, pk, pklen, NULL, cl, db_version, site_id, site_len, seq, rowid);
1788 58 : }
1789 :
1790 : // MARK: - Block-level merge helpers -
1791 :
1792 : // Store a block value in the blocks table
1793 379 : static int block_store_value (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *block_colname, dbvalue_t *col_value) {
1794 379 : dbvm_t *vm = table->block_value_write_stmt;
1795 379 : if (!vm) return cloudsync_set_error(data, "block_store_value: blocks table not initialized", DBRES_MISUSE);
1796 :
1797 379 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
1798 379 : if (rc != DBRES_OK) goto cleanup;
1799 379 : rc = databasevm_bind_text(vm, 2, block_colname, -1);
1800 379 : if (rc != DBRES_OK) goto cleanup;
1801 379 : if (col_value) {
1802 379 : rc = databasevm_bind_value(vm, 3, col_value);
1803 379 : } else {
1804 0 : rc = databasevm_bind_null(vm, 3);
1805 : }
1806 379 : if (rc != DBRES_OK) goto cleanup;
1807 :
1808 379 : rc = databasevm_step(vm);
1809 379 : if (rc == DBRES_DONE) rc = DBRES_OK;
1810 :
1811 : cleanup:
1812 379 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1813 379 : databasevm_reset(vm);
1814 379 : return rc;
1815 379 : }
1816 :
1817 : // Delete a block value from the blocks table
1818 74 : static int block_delete_value (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *block_colname) {
1819 74 : dbvm_t *vm = table->block_value_delete_stmt;
1820 74 : if (!vm) return cloudsync_set_error(data, "block_delete_value: blocks table not initialized", DBRES_MISUSE);
1821 :
1822 74 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
1823 74 : if (rc != DBRES_OK) goto cleanup;
1824 74 : rc = databasevm_bind_text(vm, 2, block_colname, -1);
1825 74 : if (rc != DBRES_OK) goto cleanup;
1826 :
1827 74 : rc = databasevm_step(vm);
1828 74 : if (rc == DBRES_DONE) rc = DBRES_OK;
1829 :
1830 : cleanup:
1831 74 : if (rc != DBRES_OK) cloudsync_set_dberror(data);
1832 74 : databasevm_reset(vm);
1833 74 : return rc;
1834 74 : }
1835 :
1836 : // Materialize all alive blocks for a base column into the base table
1837 462 : int block_materialize_column (cloudsync_context *data, cloudsync_table_context *table, const void *pk, int pklen, const char *base_col_name) {
1838 462 : if (!table->block_list_stmt) return cloudsync_set_error(data, "block_materialize_column: blocks table not initialized", DBRES_MISUSE);
1839 :
1840 : // Find column index and delimiter
1841 462 : int col_idx = -1;
1842 468 : for (int i = 0; i < table->ncols; i++) {
1843 468 : if (strcasecmp(table->col_name[i], base_col_name) == 0) {
1844 462 : col_idx = i;
1845 462 : break;
1846 : }
1847 6 : }
1848 462 : if (col_idx < 0) return cloudsync_set_error(data, "block_materialize_column: column not found", DBRES_ERROR);
1849 462 : const char *delimiter = table->col_delimiter[col_idx] ? table->col_delimiter[col_idx] : BLOCK_DEFAULT_DELIMITER;
1850 :
1851 : // Build the LIKE pattern for block col_names: "base_col\x1F%"
1852 462 : char *like_pattern = block_build_colname(base_col_name, "%");
1853 462 : if (!like_pattern) return DBRES_NOMEM;
1854 :
1855 : // Query alive blocks from blocks table joined with metadata
1856 : // block_list_stmt: SELECT b.col_value FROM blocks b JOIN meta m
1857 : // ON b.pk = m.pk AND b.col_name = m.col_name
1858 : // WHERE b.pk = ? AND b.col_name LIKE ? AND m.col_version % 2 = 1
1859 : // ORDER BY b.col_name
1860 462 : dbvm_t *vm = table->block_list_stmt;
1861 462 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
1862 462 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_reset(vm); return rc; }
1863 462 : rc = databasevm_bind_text(vm, 2, like_pattern, -1);
1864 462 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_reset(vm); return rc; }
1865 : // Bind pk again for the join condition (parameter 3)
1866 462 : rc = databasevm_bind_blob(vm, 3, pk, pklen);
1867 462 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_reset(vm); return rc; }
1868 462 : rc = databasevm_bind_text(vm, 4, like_pattern, -1);
1869 462 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_reset(vm); return rc; }
1870 :
1871 : // Collect block values
1872 462 : const char **block_values = NULL;
1873 462 : int block_count = 0;
1874 462 : int block_cap = 0;
1875 :
1876 22695 : while ((rc = databasevm_step(vm)) == DBRES_ROW) {
1877 22233 : const char *value = database_column_text(vm, 0);
1878 22233 : if (block_count >= block_cap) {
1879 1076 : int new_cap = block_cap ? block_cap * 2 : 16;
1880 1076 : const char **new_arr = (const char **)cloudsync_memory_realloc((void *)block_values, (uint64_t)(new_cap * sizeof(char *)));
1881 1076 : if (!new_arr) { rc = DBRES_NOMEM; break; }
1882 1076 : block_values = new_arr;
1883 1076 : block_cap = new_cap;
1884 1076 : }
1885 22233 : block_values[block_count] = value ? cloudsync_string_dup(value) : cloudsync_string_dup("");
1886 22233 : block_count++;
1887 : }
1888 462 : databasevm_reset(vm);
1889 462 : cloudsync_memory_free(like_pattern);
1890 :
1891 462 : if (rc != DBRES_DONE && rc != DBRES_OK && rc != DBRES_ROW) {
1892 : // Free collected values
1893 0 : for (int i = 0; i < block_count; i++) cloudsync_memory_free((void *)block_values[i]);
1894 0 : if (block_values) cloudsync_memory_free((void *)block_values);
1895 0 : return cloudsync_set_dberror(data);
1896 : }
1897 :
1898 : // Materialize text (NULL when no alive blocks)
1899 462 : char *text = (block_count > 0) ? block_materialize_text(block_values, block_count, delimiter) : NULL;
1900 22695 : for (int i = 0; i < block_count; i++) cloudsync_memory_free((void *)block_values[i]);
1901 462 : if (block_values) cloudsync_memory_free((void *)block_values);
1902 462 : if (block_count > 0 && !text) return DBRES_NOMEM;
1903 :
1904 : // Update the base table column via the col_merge_stmt (with triggers disabled)
1905 462 : dbvm_t *merge_vm = table->col_merge_stmt[col_idx];
1906 462 : if (!merge_vm) { cloudsync_memory_free(text); return DBRES_ERROR; }
1907 :
1908 : // Bind PKs
1909 462 : rc = pk_decode_prikey((char *)pk, (size_t)pklen, pk_decode_bind_callback, merge_vm);
1910 462 : if (rc < 0) { cloudsync_memory_free(text); databasevm_reset(merge_vm); return DBRES_ERROR; }
1911 :
1912 : // Bind the text value twice (INSERT value + ON CONFLICT UPDATE value)
1913 462 : int npks = table->npks;
1914 462 : if (text) {
1915 458 : rc = databasevm_bind_text(merge_vm, npks + 1, text, -1);
1916 458 : if (rc != DBRES_OK) { cloudsync_memory_free(text); databasevm_reset(merge_vm); return rc; }
1917 458 : rc = databasevm_bind_text(merge_vm, npks + 2, text, -1);
1918 458 : if (rc != DBRES_OK) { cloudsync_memory_free(text); databasevm_reset(merge_vm); return rc; }
1919 458 : } else {
1920 4 : rc = databasevm_bind_null(merge_vm, npks + 1);
1921 4 : if (rc != DBRES_OK) { databasevm_reset(merge_vm); return rc; }
1922 4 : rc = databasevm_bind_null(merge_vm, npks + 2);
1923 4 : if (rc != DBRES_OK) { databasevm_reset(merge_vm); return rc; }
1924 : }
1925 :
1926 : // Execute with triggers disabled
1927 462 : table->enabled = 0;
1928 462 : SYNCBIT_SET(data);
1929 462 : rc = databasevm_step(merge_vm);
1930 462 : databasevm_reset(merge_vm);
1931 462 : SYNCBIT_RESET(data);
1932 462 : table->enabled = 1;
1933 :
1934 462 : cloudsync_memory_free(text);
1935 :
1936 462 : if (rc == DBRES_DONE) rc = DBRES_OK;
1937 462 : if (rc != DBRES_OK) return cloudsync_set_dberror(data);
1938 462 : return DBRES_OK;
1939 462 : }
1940 :
1941 : // Accessor for has_block_cols flag
1942 1024 : bool table_has_block_cols (cloudsync_table_context *table) {
1943 1024 : return table && table->has_block_cols;
1944 : }
1945 :
1946 : // Get block column algo for a given column index
1947 13128 : col_algo_t table_col_algo (cloudsync_table_context *table, int index) {
1948 13128 : if (!table || !table->col_algo || index < 0 || index >= table->ncols) return col_algo_normal;
1949 13128 : return table->col_algo[index];
1950 13128 : }
1951 :
1952 : // Get block delimiter for a given column index
1953 137 : const char *table_col_delimiter (cloudsync_table_context *table, int index) {
1954 137 : if (!table || !table->col_delimiter || index < 0 || index >= table->ncols) return BLOCK_DEFAULT_DELIMITER;
1955 137 : return table->col_delimiter[index] ? table->col_delimiter[index] : BLOCK_DEFAULT_DELIMITER;
1956 137 : }
1957 :
1958 : // Block column struct accessors (for use outside cloudsync.c where struct is opaque)
1959 1024 : dbvm_t *table_block_value_read_stmt (cloudsync_table_context *table) { return table ? table->block_value_read_stmt : NULL; }
1960 506 : dbvm_t *table_block_value_write_stmt (cloudsync_table_context *table) { return table ? table->block_value_write_stmt : NULL; }
1961 93 : dbvm_t *table_block_list_stmt (cloudsync_table_context *table) { return table ? table->block_list_stmt : NULL; }
1962 93 : const char *table_blocks_ref (cloudsync_table_context *table) { return table ? table->blocks_ref : NULL; }
1963 :
1964 3 : void table_set_col_delimiter (cloudsync_table_context *table, int col_idx, const char *delimiter) {
1965 3 : if (!table || !table->col_delimiter || col_idx < 0 || col_idx >= table->ncols) return;
1966 3 : if (table->col_delimiter[col_idx]) cloudsync_memory_free(table->col_delimiter[col_idx]);
1967 3 : table->col_delimiter[col_idx] = delimiter ? cloudsync_string_dup(delimiter) : NULL;
1968 3 : }
1969 :
1970 : // Find column index by name
1971 127 : int table_col_index (cloudsync_table_context *table, const char *col_name) {
1972 127 : if (!table || !col_name) return -1;
1973 131 : for (int i = 0; i < table->ncols; i++) {
1974 131 : if (strcasecmp(table->col_name[i], col_name) == 0) return i;
1975 4 : }
1976 0 : return -1;
1977 127 : }
1978 :
1979 48345 : int merge_insert (cloudsync_context *data, cloudsync_table_context *table, const char *insert_pk, int insert_pk_len, int64_t insert_cl, const char *insert_name, dbvalue_t *insert_value, int64_t insert_col_version, int64_t insert_db_version, const char *insert_site_id, int insert_site_id_len, int64_t insert_seq, int64_t *rowid) {
1980 : // Handle DWS and AWS algorithms here
1981 : // Delete-Wins Set (DWS): table_algo_crdt_dws
1982 : // Add-Wins Set (AWS): table_algo_crdt_aws
1983 :
1984 : // Causal-Length Set (CLS) Algorithm (default)
1985 :
1986 : // compute the local causal length for the row based on the primary key
1987 : // the causal length is used to determine the order of operations and resolve conflicts.
1988 48345 : int64_t local_cl = merge_get_local_cl(table, insert_pk, insert_pk_len);
1989 48345 : if (local_cl < 0) return cloudsync_set_error(data, "Unable to compute local causal length", DBRES_ERROR);
1990 :
1991 : // if the incoming causal length is older than the local causal length, we can safely ignore it
1992 : // because the local changes are more recent
1993 48345 : if (insert_cl < local_cl) return DBRES_OK;
1994 :
1995 : // check if the operation is a delete by examining the causal length
1996 : // even causal lengths typically signify delete operations
1997 48128 : bool is_delete = (insert_cl % 2 == 0);
1998 48128 : if (is_delete) {
1999 : // if it's a delete, check if the local state is at the same causal length
2000 : // if it is, no further action is needed
2001 610 : if (local_cl == insert_cl) return DBRES_OK;
2002 :
2003 : // perform a delete merge if the causal length is newer than the local one
2004 334 : int rc = merge_delete(data, table, insert_pk, insert_pk_len, insert_name, insert_col_version,
2005 167 : insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2006 167 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_delete", rc);
2007 167 : return rc;
2008 : }
2009 :
2010 : // if the operation is a sentinel-only insert (indicating a new row or resurrected row with no column update), handle it separately.
2011 47518 : bool is_sentinel_only = (strcmp(insert_name, CLOUDSYNC_TOMBSTONE_VALUE) == 0);
2012 47518 : if (is_sentinel_only) {
2013 185 : if (local_cl == insert_cl) return DBRES_OK;
2014 :
2015 : // perform a sentinel-only insert to track the existence of the row
2016 116 : int rc = merge_sentinel_only_insert(data, table, insert_pk, insert_pk_len, insert_col_version,
2017 58 : insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2018 58 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_sentinel_only_insert", rc);
2019 58 : return rc;
2020 : }
2021 :
2022 : // from this point I can be sure that insert_name is not sentinel
2023 :
2024 : // handle the case where a row is being resurrected (e.g., after a delete, a new insert for the same row)
2025 : // odd causal lengths can "resurrect" rows
2026 47333 : bool needs_resurrect = (insert_cl > local_cl && insert_cl % 2 == 1);
2027 47333 : bool row_exists_locally = local_cl != 0;
2028 :
2029 : // if a resurrection is needed, insert a sentinel to mark the row as alive
2030 : // this handles out-of-order deliveries where the row was deleted and is now being re-inserted
2031 47333 : if (needs_resurrect && (row_exists_locally || (!row_exists_locally && insert_cl > 1))) {
2032 0 : int rc = merge_sentinel_only_insert(data, table, insert_pk, insert_pk_len, insert_cl,
2033 0 : insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2034 0 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to perform merge_sentinel_only_insert", rc);
2035 0 : }
2036 :
2037 : // at this point, we determine whether the incoming change wins based on causal length
2038 : // this can be due to a resurrection, a non-existent local row, or a conflict resolution
2039 47333 : bool flag = false;
2040 47333 : int rc = merge_did_cid_win(data, table, insert_pk, insert_pk_len, insert_value, insert_site_id, insert_site_id_len, insert_name, insert_col_version, &flag);
2041 47333 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to perform merge_did_cid_win", rc);
2042 :
2043 : // check if the incoming change wins and should be applied
2044 47333 : bool does_cid_win = ((needs_resurrect) || (!row_exists_locally) || (flag));
2045 47333 : if (!does_cid_win) return DBRES_OK;
2046 :
2047 : // Block-level LWW: if the incoming col_name is a block entry (contains \x1F),
2048 : // bypass the normal base-table write and instead store the value in the blocks table.
2049 : // The base table column will be materialized from all alive blocks.
2050 23868 : if (block_is_block_colname(insert_name) && table->has_block_cols) {
2051 : // Store or delete block value in blocks table depending on tombstone status
2052 412 : if (insert_col_version % 2 == 0) {
2053 : // Tombstone: remove from blocks table
2054 33 : rc = block_delete_value(data, table, insert_pk, insert_pk_len, insert_name);
2055 33 : } else {
2056 379 : rc = block_store_value(data, table, insert_pk, insert_pk_len, insert_name, insert_value);
2057 : }
2058 412 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to store/delete block value", rc);
2059 :
2060 : // Set winner clock in metadata
2061 824 : rc = merge_set_winner_clock(data, table, insert_pk, insert_pk_len, insert_name,
2062 412 : insert_col_version, insert_db_version,
2063 412 : insert_site_id, insert_site_id_len, insert_seq, rowid);
2064 412 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to set winner clock for block", rc);
2065 :
2066 : // Materialize the full column from blocks into the base table
2067 412 : char *base_col = block_extract_base_colname(insert_name);
2068 412 : if (base_col) {
2069 412 : rc = block_materialize_column(data, table, insert_pk, insert_pk_len, base_col);
2070 412 : cloudsync_memory_free(base_col);
2071 412 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to materialize block column", rc);
2072 412 : }
2073 :
2074 412 : return DBRES_OK;
2075 : }
2076 :
2077 : // perform the final column insert or update if the incoming change wins
2078 23456 : if (data->pending_batch) {
2079 : // Propagate row_exists_locally to the batch on the first winning column.
2080 : // This lets merge_flush_pending choose UPDATE vs INSERT ON CONFLICT,
2081 : // which matters when RLS policies reference columns not in the payload.
2082 22939 : if (data->pending_batch->table == NULL) {
2083 8760 : data->pending_batch->row_exists = row_exists_locally;
2084 8760 : }
2085 22939 : rc = merge_pending_add(data, table, insert_pk, insert_pk_len, insert_name, insert_value, insert_col_version, insert_db_version, insert_site_id, insert_site_id_len, insert_seq);
2086 22939 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_pending_add", rc);
2087 22939 : } else {
2088 517 : rc = merge_insert_col(data, table, insert_pk, insert_pk_len, insert_name, insert_value, insert_col_version, insert_db_version, insert_site_id, insert_site_id_len, insert_seq, rowid);
2089 517 : if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to perform merge_insert_col", rc);
2090 : }
2091 :
2092 23456 : return rc;
2093 48345 : }
2094 :
2095 : // MARK: - Block column setup -
2096 :
2097 : // Migrate existing tracked rows to block format when block-level LWW is first enabled on a column.
2098 : // Scans the metadata table for alive rows with the plain col_name entry (not yet block entries),
2099 : // reads each row's current value from the base table, splits it into blocks, and inserts
2100 : // the block entries into both the blocks table and the metadata table.
2101 : // Uses INSERT OR IGNORE semantics so the operation is safe to call multiple times.
2102 73 : static int block_migrate_existing_rows (cloudsync_context *data, cloudsync_table_context *table, int col_idx) {
2103 73 : const char *col_name = table->col_name[col_idx];
2104 73 : if (!col_name || !table->meta_ref || !table->blocks_ref) return DBRES_OK;
2105 :
2106 73 : const char *delim = table->col_delimiter[col_idx] ? table->col_delimiter[col_idx] : BLOCK_DEFAULT_DELIMITER;
2107 73 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
2108 :
2109 : // Phase 1: collect all existing PKs that have an alive regular col_name entry
2110 : // AND do not yet have any entries in the blocks table for this column.
2111 : // The NOT IN filter makes this idempotent: rows that were already migrated
2112 : // (or had their blocks created via INSERT) are skipped on subsequent calls.
2113 : // We collect PKs before writing so that writes to the metadata table (Phase 2)
2114 : // do not perturb the read cursor on the same table.
2115 73 : char *like_pattern = block_build_colname(col_name, "%");
2116 73 : if (!like_pattern) return DBRES_NOMEM;
2117 :
2118 73 : char *scan_sql = cloudsync_memory_mprintf(SQL_META_SCAN_COL_FOR_MIGRATION, table->meta_ref, table->blocks_ref);
2119 73 : if (!scan_sql) { cloudsync_memory_free(like_pattern); return DBRES_NOMEM; }
2120 73 : dbvm_t *scan_vm = NULL;
2121 73 : int rc = databasevm_prepare(data, scan_sql, &scan_vm, 0);
2122 73 : cloudsync_memory_free(scan_sql);
2123 73 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); return rc; }
2124 :
2125 73 : rc = databasevm_bind_text(scan_vm, 1, col_name, -1);
2126 73 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_finalize(scan_vm); return rc; }
2127 : // Bind like_pattern as ?2 and keep it alive until after all scan steps complete,
2128 : // because databasevm_bind_text uses SQLITE_STATIC (no copy).
2129 73 : rc = databasevm_bind_text(scan_vm, 2, like_pattern, -1);
2130 73 : if (rc != DBRES_OK) { cloudsync_memory_free(like_pattern); databasevm_finalize(scan_vm); return rc; }
2131 :
2132 : // Collect pk blobs into a dynamically-grown array of owned copies
2133 73 : void **pks = NULL;
2134 73 : size_t *pklens = NULL;
2135 73 : int pk_count = 0;
2136 73 : int pk_cap = 0;
2137 :
2138 75 : while ((rc = databasevm_step(scan_vm)) == DBRES_ROW) {
2139 2 : size_t pklen = 0;
2140 2 : const void *pk = database_column_blob(scan_vm, 0, &pklen);
2141 2 : if (!pk || pklen == 0) continue;
2142 :
2143 2 : if (pk_count >= pk_cap) {
2144 1 : int new_cap = pk_cap ? pk_cap * 2 : 8;
2145 1 : void **new_pks = (void **)cloudsync_memory_realloc(pks, (uint64_t)(new_cap * sizeof(void *)));
2146 1 : size_t *new_pklens = (size_t *)cloudsync_memory_realloc(pklens, (uint64_t)(new_cap * sizeof(size_t)));
2147 1 : if (!new_pks || !new_pklens) {
2148 0 : cloudsync_memory_free(new_pks ? new_pks : pks);
2149 0 : cloudsync_memory_free(new_pklens ? new_pklens : pklens);
2150 0 : databasevm_finalize(scan_vm);
2151 0 : return DBRES_NOMEM;
2152 : }
2153 1 : pks = new_pks;
2154 1 : pklens = new_pklens;
2155 1 : pk_cap = new_cap;
2156 1 : }
2157 :
2158 2 : pks[pk_count] = cloudsync_memory_alloc((uint64_t)pklen);
2159 2 : if (!pks[pk_count]) { rc = DBRES_NOMEM; break; }
2160 2 : memcpy(pks[pk_count], pk, pklen);
2161 2 : pklens[pk_count] = pklen;
2162 2 : pk_count++;
2163 : }
2164 :
2165 73 : databasevm_finalize(scan_vm);
2166 73 : cloudsync_memory_free(like_pattern); // safe to free after scan_vm is finalized
2167 73 : if (rc != DBRES_DONE && rc != DBRES_OK) {
2168 0 : for (int i = 0; i < pk_count; i++) cloudsync_memory_free(pks[i]);
2169 0 : cloudsync_memory_free(pks);
2170 0 : cloudsync_memory_free(pklens);
2171 0 : return rc;
2172 : }
2173 :
2174 73 : if (pk_count == 0) {
2175 72 : cloudsync_memory_free(pks);
2176 72 : cloudsync_memory_free(pklens);
2177 72 : return DBRES_OK;
2178 : }
2179 :
2180 : // Phase 2: for each collected PK, read the column value, split into blocks,
2181 : // and insert into the blocks table + metadata using INSERT OR IGNORE.
2182 :
2183 1 : char *meta_sql = cloudsync_memory_mprintf(SQL_META_INSERT_BLOCK_IGNORE, table->meta_ref);
2184 1 : if (!meta_sql) { rc = DBRES_NOMEM; goto cleanup_pks; }
2185 1 : dbvm_t *meta_vm = NULL;
2186 1 : rc = databasevm_prepare(data, meta_sql, &meta_vm, 0);
2187 1 : cloudsync_memory_free(meta_sql);
2188 1 : if (rc != DBRES_OK) goto cleanup_pks;
2189 :
2190 1 : char *blocks_sql = cloudsync_memory_mprintf(SQL_BLOCKS_INSERT_IGNORE, table->blocks_ref);
2191 1 : if (!blocks_sql) { databasevm_finalize(meta_vm); rc = DBRES_NOMEM; goto cleanup_pks; }
2192 1 : dbvm_t *blocks_vm = NULL;
2193 1 : rc = databasevm_prepare(data, blocks_sql, &blocks_vm, 0);
2194 1 : cloudsync_memory_free(blocks_sql);
2195 1 : if (rc != DBRES_OK) { databasevm_finalize(meta_vm); goto cleanup_pks; }
2196 :
2197 1 : dbvm_t *val_vm = (dbvm_t *)table_column_lookup(table, col_name, false, NULL);
2198 :
2199 3 : for (int p = 0; p < pk_count; p++) {
2200 2 : const void *pk = pks[p];
2201 2 : size_t pklen = pklens[p];
2202 :
2203 2 : if (!val_vm) continue;
2204 :
2205 : // Read current column value from the base table
2206 2 : int bind_rc = pk_decode_prikey((char *)pk, pklen, pk_decode_bind_callback, (void *)val_vm);
2207 2 : if (bind_rc < 0) { databasevm_reset(val_vm); continue; }
2208 :
2209 2 : int step_rc = databasevm_step(val_vm);
2210 2 : const char *text = (step_rc == DBRES_ROW) ? database_column_text(val_vm, 0) : NULL;
2211 : // Make a copy of text before resetting val_vm, as the pointer is only valid until reset
2212 2 : char *text_copy = text ? cloudsync_string_dup(text) : NULL;
2213 2 : databasevm_reset(val_vm);
2214 :
2215 2 : if (!text_copy) continue; // NULL column value: nothing to migrate
2216 :
2217 : // Split text into blocks and store each one
2218 2 : block_list_t *blocks = block_split(text_copy, delim);
2219 2 : cloudsync_memory_free(text_copy);
2220 2 : if (!blocks) continue;
2221 :
2222 2 : char **positions = block_initial_positions(blocks->count);
2223 2 : if (positions) {
2224 7 : for (int b = 0; b < blocks->count; b++) {
2225 5 : char *block_cn = block_build_colname(col_name, positions[b]);
2226 5 : if (block_cn) {
2227 : // Metadata entry (skip if this block position already exists)
2228 5 : databasevm_bind_blob(meta_vm, 1, pk, (int)pklen);
2229 5 : databasevm_bind_text(meta_vm, 2, block_cn, -1);
2230 5 : databasevm_bind_int(meta_vm, 3, 1); // col_version = 1 (alive)
2231 5 : databasevm_bind_int(meta_vm, 4, db_version);
2232 5 : databasevm_bind_int(meta_vm, 5, cloudsync_bumpseq(data));
2233 5 : databasevm_step(meta_vm);
2234 5 : databasevm_reset(meta_vm);
2235 :
2236 : // Block value (skip if this block position already exists)
2237 5 : databasevm_bind_blob(blocks_vm, 1, pk, (int)pklen);
2238 5 : databasevm_bind_text(blocks_vm, 2, block_cn, -1);
2239 5 : databasevm_bind_text(blocks_vm, 3, blocks->entries[b].content, -1);
2240 5 : databasevm_step(blocks_vm);
2241 5 : databasevm_reset(blocks_vm);
2242 :
2243 5 : cloudsync_memory_free(block_cn);
2244 5 : }
2245 5 : cloudsync_memory_free(positions[b]);
2246 5 : }
2247 2 : cloudsync_memory_free(positions);
2248 2 : }
2249 2 : block_list_free(blocks);
2250 2 : }
2251 :
2252 1 : databasevm_finalize(meta_vm);
2253 1 : databasevm_finalize(blocks_vm);
2254 1 : rc = DBRES_OK;
2255 :
2256 : cleanup_pks:
2257 3 : for (int i = 0; i < pk_count; i++) cloudsync_memory_free(pks[i]);
2258 1 : cloudsync_memory_free(pks);
2259 1 : cloudsync_memory_free(pklens);
2260 1 : return rc;
2261 73 : }
2262 :
2263 74 : int cloudsync_setup_block_column (cloudsync_context *data, const char *table_name, const char *col_name, const char *delimiter, bool persist) {
2264 74 : cloudsync_table_context *table = table_lookup(data, table_name);
2265 74 : if (!table) return cloudsync_set_error(data, "cloudsync_setup_block_column: table not found", DBRES_ERROR);
2266 :
2267 : // Find column index
2268 74 : int col_idx = table_col_index(table, col_name);
2269 74 : if (col_idx < 0) {
2270 : char buf[1024];
2271 0 : snprintf(buf, sizeof(buf), "cloudsync_setup_block_column: column '%s' not found in table '%s'", col_name, table_name);
2272 0 : return cloudsync_set_error(data, buf, DBRES_ERROR);
2273 : }
2274 :
2275 : // Set column algo
2276 74 : table->col_algo[col_idx] = col_algo_block;
2277 74 : table->has_block_cols = true;
2278 :
2279 : // Set delimiter (can be NULL for default)
2280 74 : if (table->col_delimiter[col_idx]) {
2281 0 : cloudsync_memory_free(table->col_delimiter[col_idx]);
2282 0 : table->col_delimiter[col_idx] = NULL;
2283 0 : }
2284 74 : if (delimiter) {
2285 1 : table->col_delimiter[col_idx] = cloudsync_string_dup(delimiter);
2286 1 : }
2287 :
2288 : // Create blocks table if not already done
2289 74 : if (!table->blocks_ref) {
2290 71 : table->blocks_ref = database_build_blocks_ref(table->schema, table->name);
2291 71 : if (!table->blocks_ref) return DBRES_NOMEM;
2292 :
2293 : // CREATE TABLE IF NOT EXISTS
2294 71 : char *sql = cloudsync_memory_mprintf(SQL_BLOCKS_CREATE_TABLE, table->blocks_ref);
2295 71 : if (!sql) return DBRES_NOMEM;
2296 :
2297 71 : int rc = database_exec(data, sql);
2298 71 : cloudsync_memory_free(sql);
2299 71 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Unable to create blocks table", rc);
2300 :
2301 : // Prepare block statements
2302 : // Write: upsert into blocks (pk, col_name, col_value)
2303 71 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_UPSERT, table->blocks_ref);
2304 71 : if (!sql) return DBRES_NOMEM;
2305 71 : rc = databasevm_prepare(data, sql, (void **)&table->block_value_write_stmt, DBFLAG_PERSISTENT);
2306 71 : cloudsync_memory_free(sql);
2307 71 : if (rc != DBRES_OK) return rc;
2308 :
2309 : // Read: SELECT col_value FROM blocks WHERE pk = ? AND col_name = ?
2310 71 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_SELECT, table->blocks_ref);
2311 71 : if (!sql) return DBRES_NOMEM;
2312 71 : rc = databasevm_prepare(data, sql, (void **)&table->block_value_read_stmt, DBFLAG_PERSISTENT);
2313 71 : cloudsync_memory_free(sql);
2314 71 : if (rc != DBRES_OK) return rc;
2315 :
2316 : // Delete: DELETE FROM blocks WHERE pk = ? AND col_name = ?
2317 71 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_DELETE, table->blocks_ref);
2318 71 : if (!sql) return DBRES_NOMEM;
2319 71 : rc = databasevm_prepare(data, sql, (void **)&table->block_value_delete_stmt, DBFLAG_PERSISTENT);
2320 71 : cloudsync_memory_free(sql);
2321 71 : if (rc != DBRES_OK) return rc;
2322 :
2323 : // List alive blocks for materialization
2324 71 : sql = cloudsync_memory_mprintf(SQL_BLOCKS_LIST_ALIVE, table->blocks_ref, table->meta_ref);
2325 71 : if (!sql) return DBRES_NOMEM;
2326 71 : rc = databasevm_prepare(data, sql, (void **)&table->block_list_stmt, DBFLAG_PERSISTENT);
2327 71 : cloudsync_memory_free(sql);
2328 71 : if (rc != DBRES_OK) return rc;
2329 71 : }
2330 :
2331 : // Persist settings (skipped when called from the settings loader, since
2332 : // writing to cloudsync_table_settings while sqlite3_exec is iterating it
2333 : // re-feeds the rewritten row to the cursor and causes an infinite loop).
2334 74 : if (persist) {
2335 73 : int rc = dbutils_table_settings_set_key_value(data, table_name, col_name, "algo", "block");
2336 73 : if (rc != DBRES_OK) return rc;
2337 :
2338 73 : if (delimiter) {
2339 0 : rc = dbutils_table_settings_set_key_value(data, table_name, col_name, "delimiter", delimiter);
2340 0 : if (rc != DBRES_OK) return rc;
2341 0 : }
2342 :
2343 : // Migrate any existing tracked rows: populate the blocks table and metadata with
2344 : // block entries derived from the current column value, so that subsequent UPDATE
2345 : // operations can diff against the real existing state instead of treating everything
2346 : // as new, and so this node participates correctly in LWW conflict resolution.
2347 73 : rc = block_migrate_existing_rows(data, table, col_idx);
2348 73 : if (rc != DBRES_OK) return rc;
2349 73 : }
2350 :
2351 74 : return DBRES_OK;
2352 74 : }
2353 :
2354 : // MARK: - Private -
2355 :
2356 255 : bool cloudsync_config_exists (cloudsync_context *data) {
2357 255 : return database_internal_table_exists(data, CLOUDSYNC_SITEID_NAME) == true;
2358 : }
2359 :
2360 801 : bool cloudsync_context_is_initialized (cloudsync_context *data) {
2361 : // A fully initialized context has its persistent "is the DB stale" probe
2362 : // prepared. cloudsync_context_init prepares data_version_stmt (via
2363 : // cloudsync_add_dbvms) only after the cloudsync_site_id table exists, so
2364 : // a non-NULL pointer means cloudsync_init has been called at least once
2365 : // on this connection. Used to produce actionable error messages when
2366 : // callers hit a function before calling cloudsync_init.
2367 801 : return data != NULL && data->data_version_stmt != NULL;
2368 : }
2369 :
2370 270 : cloudsync_context *cloudsync_context_create (void *db) {
2371 270 : cloudsync_context *data = (cloudsync_context *)cloudsync_memory_zeroalloc((uint64_t)(sizeof(cloudsync_context)));
2372 270 : if (!data) return NULL;
2373 : DEBUG_SETTINGS("cloudsync_context_create %p", data);
2374 :
2375 270 : data->libversion = CLOUDSYNC_VERSION;
2376 270 : data->pending_db_version = CLOUDSYNC_VALUE_NOTSET;
2377 : #if CLOUDSYNC_DEBUG
2378 : data->debug = 1;
2379 : #endif
2380 :
2381 : // allocate space for 64 tables (it can grow if needed)
2382 270 : uint64_t mem_needed = (uint64_t)(CLOUDSYNC_INIT_NTABLES * sizeof(cloudsync_table_context *));
2383 270 : data->tables = (cloudsync_table_context **)cloudsync_memory_zeroalloc(mem_needed);
2384 270 : if (!data->tables) {cloudsync_memory_free(data); return NULL;}
2385 :
2386 270 : data->tables_cap = CLOUDSYNC_INIT_NTABLES;
2387 270 : data->tables_count = 0;
2388 270 : data->db = db;
2389 :
2390 : // SQLite exposes col_value as ANY, but other databases require a concrete type.
2391 : // In PostgreSQL we expose col_value as bytea, which holds the pk-encoded value bytes (type + data).
2392 : // Because col_value is already encoded, we skip decoding this field and pass it through as bytea.
2393 : // It is decoded to the target column type just before applying changes to the base table.
2394 270 : data->skip_decode_idx = (db == NULL) ? CLOUDSYNC_PK_INDEX_COLVALUE : -1;
2395 :
2396 270 : return data;
2397 270 : }
2398 :
2399 270 : void cloudsync_context_free (void *ctx) {
2400 270 : cloudsync_context *data = (cloudsync_context *)ctx;
2401 : DEBUG_SETTINGS("cloudsync_context_free %p", data);
2402 270 : if (!data) return;
2403 :
2404 : // free all table contexts and prepared statements
2405 270 : cloudsync_terminate(data);
2406 :
2407 270 : cloudsync_memory_free(data->tables);
2408 270 : cloudsync_memory_free(data);
2409 270 : }
2410 :
2411 364 : const char *cloudsync_context_init (cloudsync_context *data) {
2412 364 : if (!data) return NULL;
2413 :
2414 : // perform init just the first time, if the site_id field is not set.
2415 : // The data->site_id value could exists while settings tables don't exists if the
2416 : // cloudsync_context_init was previously called in init transaction that was rolled back
2417 : // because of an error during the init process.
2418 364 : if (data->site_id[0] == 0 || !database_internal_table_exists(data, CLOUDSYNC_SITEID_NAME)) {
2419 249 : if (dbutils_settings_init(data) != DBRES_OK) return NULL;
2420 249 : if (cloudsync_add_dbvms(data) != DBRES_OK) return NULL;
2421 249 : if (cloudsync_load_siteid(data) != DBRES_OK) return NULL;
2422 249 : data->schema_hash = database_schema_hash(data);
2423 249 : }
2424 :
2425 364 : return (const char *)data->site_id;
2426 364 : }
2427 :
2428 1350 : void cloudsync_sync_key (cloudsync_context *data, const char *key, const char *value) {
2429 : DEBUG_SETTINGS("cloudsync_sync_key key: %s value: %s", key, value);
2430 :
2431 : // sync data
2432 1350 : if (strcmp(key, CLOUDSYNC_KEY_SCHEMAVERSION) == 0) {
2433 249 : data->schema_version = (int)strtol(value, NULL, 0);
2434 249 : return;
2435 : }
2436 :
2437 1101 : if (strcmp(key, CLOUDSYNC_KEY_DEBUG) == 0) {
2438 0 : data->debug = 0;
2439 0 : if (value && (value[0] != 0) && (value[0] != '0')) data->debug = 1;
2440 0 : return;
2441 : }
2442 :
2443 1101 : if (strcmp(key, CLOUDSYNC_KEY_SCHEMA) == 0) {
2444 0 : cloudsync_set_schema(data, value);
2445 0 : return;
2446 : }
2447 1350 : }
2448 :
2449 : #if 0
2450 : void cloudsync_sync_table_key(cloudsync_context *data, const char *table, const char *column, const char *key, const char *value) {
2451 : DEBUG_SETTINGS("cloudsync_sync_table_key table: %s column: %s key: %s value: %s", table, column, key, value);
2452 : // Unused in this version
2453 : return;
2454 : }
2455 : #endif
2456 :
2457 8430 : int cloudsync_commit_hook (void *ctx) {
2458 8430 : cloudsync_context *data = (cloudsync_context *)ctx;
2459 :
2460 8430 : data->db_version = data->pending_db_version;
2461 8430 : data->pending_db_version = CLOUDSYNC_VALUE_NOTSET;
2462 8430 : data->seq = 0;
2463 :
2464 8430 : return DBRES_OK;
2465 : }
2466 :
2467 3 : void cloudsync_rollback_hook (void *ctx) {
2468 3 : cloudsync_context *data = (cloudsync_context *)ctx;
2469 :
2470 3 : data->pending_db_version = CLOUDSYNC_VALUE_NOTSET;
2471 3 : data->seq = 0;
2472 3 : }
2473 :
2474 24 : int cloudsync_begin_alter (cloudsync_context *data, const char *table_name) {
2475 : // init cloudsync_settings
2476 24 : if (cloudsync_context_init(data) == NULL) {
2477 0 : return DBRES_MISUSE;
2478 : }
2479 :
2480 : // lookup table
2481 24 : cloudsync_table_context *table = table_lookup(data, table_name);
2482 24 : if (!table) {
2483 : char buffer[1024];
2484 1 : snprintf(buffer, sizeof(buffer), "Unable to find table %s", table_name);
2485 1 : return cloudsync_set_error(data, buffer, DBRES_MISUSE);
2486 : }
2487 :
2488 : // idempotent: if already altering, return OK
2489 23 : if (table->is_altering) return DBRES_OK;
2490 :
2491 : // retrieve primary key(s)
2492 23 : char **names = NULL;
2493 23 : int nrows = 0;
2494 23 : int rc = database_pk_names(data, table_name, &names, &nrows);
2495 23 : if (rc != DBRES_OK) {
2496 : char buffer[1024];
2497 0 : snprintf(buffer, sizeof(buffer), "Unable to get primary keys for table %s", table_name);
2498 0 : cloudsync_set_error(data, buffer, DBRES_MISUSE);
2499 0 : goto rollback_begin_alter;
2500 : }
2501 :
2502 : // sanity check the number of primary keys
2503 23 : if (nrows != table_count_pks(table)) {
2504 : char buffer[1024];
2505 0 : snprintf(buffer, sizeof(buffer), "Number of primary keys for table %s changed before ALTER", table_name);
2506 0 : cloudsync_set_error(data, buffer, DBRES_MISUSE);
2507 0 : goto rollback_begin_alter;
2508 : }
2509 :
2510 : // drop original triggers
2511 23 : rc = database_delete_triggers(data, table_name);
2512 23 : if (rc != DBRES_OK) {
2513 : char buffer[1024];
2514 0 : snprintf(buffer, sizeof(buffer), "Unable to delete triggers for table %s in cloudsync_begin_alter.", table_name);
2515 0 : cloudsync_set_error(data, buffer, DBRES_ERROR);
2516 0 : goto rollback_begin_alter;
2517 : }
2518 :
2519 23 : table_set_pknames(table, names);
2520 23 : table->is_altering = true;
2521 23 : return DBRES_OK;
2522 :
2523 : rollback_begin_alter:
2524 0 : if (names) table_pknames_free(names, nrows);
2525 0 : return rc;
2526 24 : }
2527 :
2528 23 : int cloudsync_finalize_alter (cloudsync_context *data, cloudsync_table_context *table) {
2529 : // check if dbversion needed to be updated
2530 23 : cloudsync_dbversion_check_uptodate(data);
2531 :
2532 : // if primary-key columns change, all row identities change.
2533 : // In that case, the clock table must be dropped, recreated,
2534 : // and backfilled. We detect this by comparing the unique index
2535 : // in the lookaside table with the source table's PKs.
2536 :
2537 : // retrieve primary keys (to check is they changed)
2538 23 : char **result = NULL;
2539 23 : int nrows = 0;
2540 23 : int rc = database_pk_names (data, table->name, &result, &nrows);
2541 23 : if (rc != DBRES_OK || nrows == 0) {
2542 0 : if (nrows == 0) rc = DBRES_MISUSE;
2543 0 : goto finalize;
2544 : }
2545 :
2546 : // check if there are differences
2547 23 : bool pk_diff = (nrows != table->npks);
2548 23 : if (!pk_diff) {
2549 45 : for (int i = 0; i < nrows; ++i) {
2550 34 : if (strcmp(table->pk_name[i], result[i]) != 0) {
2551 6 : pk_diff = true;
2552 6 : break;
2553 : }
2554 28 : }
2555 17 : }
2556 :
2557 23 : if (pk_diff) {
2558 : // drop meta-table, it will be recreated
2559 12 : char *sql = cloudsync_memory_mprintf(SQL_DROP_CLOUDSYNC_TABLE, table->meta_ref);
2560 12 : rc = database_exec(data, sql);
2561 12 : cloudsync_memory_free(sql);
2562 12 : if (rc != DBRES_OK) {
2563 0 : DEBUG_DBERROR(rc, "cloudsync_finalize_alter", data);
2564 0 : goto finalize;
2565 : }
2566 12 : } else {
2567 : // compact meta-table
2568 : // delete entries for removed columns
2569 11 : const char *schema = table->schema ? table->schema : "";
2570 11 : char *sql = sql_build_delete_cols_not_in_schema_query(schema, table->name, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
2571 11 : rc = database_exec(data, sql);
2572 11 : cloudsync_memory_free(sql);
2573 11 : if (rc != DBRES_OK) {
2574 0 : DEBUG_DBERROR(rc, "cloudsync_finalize_alter", data);
2575 0 : goto finalize;
2576 : }
2577 :
2578 11 : sql = sql_build_pk_qualified_collist_query(schema, table->name);
2579 11 : if (!sql) {rc = DBRES_NOMEM; goto finalize;}
2580 :
2581 11 : char *pkclause = NULL;
2582 11 : rc = database_select_text(data, sql, &pkclause);
2583 11 : cloudsync_memory_free(sql);
2584 11 : if (rc != DBRES_OK) goto finalize;
2585 11 : char *pkvalues = (pkclause) ? pkclause : "rowid";
2586 :
2587 : // delete entries related to rows that no longer exist in the original table, but preserve tombstone
2588 11 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_GC_DELETE_ORPHANED_PK, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE, CLOUDSYNC_TOMBSTONE_VALUE, table->base_ref, table->meta_ref, pkvalues);
2589 11 : rc = database_exec(data, sql);
2590 11 : if (pkclause) cloudsync_memory_free(pkclause);
2591 11 : cloudsync_memory_free(sql);
2592 11 : if (rc != DBRES_OK) {
2593 0 : DEBUG_DBERROR(rc, "cloudsync_finalize_alter", data);
2594 0 : goto finalize;
2595 : }
2596 :
2597 : }
2598 :
2599 : // update key to be later used in cloudsync_dbversion_rebuild
2600 : char buf[256];
2601 23 : snprintf(buf, sizeof(buf), "%" PRId64, data->db_version);
2602 23 : dbutils_settings_set_key_value(data, "pre_alter_dbversion", buf);
2603 :
2604 : finalize:
2605 23 : table_pknames_free(result, nrows);
2606 23 : return rc;
2607 : }
2608 :
2609 24 : int cloudsync_commit_alter (cloudsync_context *data, const char *table_name) {
2610 24 : int rc = DBRES_MISUSE;
2611 24 : cloudsync_table_context *table = NULL;
2612 :
2613 : // init cloudsync_settings
2614 24 : if (cloudsync_context_init(data) == NULL) {
2615 0 : cloudsync_set_error(data, "Unable to initialize cloudsync context", DBRES_MISUSE);
2616 0 : goto rollback_finalize_alter;
2617 : }
2618 :
2619 : // lookup table
2620 24 : table = table_lookup(data, table_name);
2621 24 : if (!table) {
2622 : char buffer[1024];
2623 1 : snprintf(buffer, sizeof(buffer), "Unable to find table %s", table_name);
2624 1 : cloudsync_set_error(data, buffer, DBRES_MISUSE);
2625 1 : goto rollback_finalize_alter;
2626 : }
2627 :
2628 : // idempotent: if not altering, return OK
2629 23 : if (!table->is_altering) return DBRES_OK;
2630 :
2631 23 : rc = cloudsync_finalize_alter(data, table);
2632 23 : if (rc != DBRES_OK) goto rollback_finalize_alter;
2633 :
2634 : // the table is outdated, delete it and it will be reloaded in the cloudsync_init_internal
2635 : // is_altering is reset implicitly because table_free + cloudsync_init_table
2636 : // will reallocate the table context with zero-initialized memory
2637 23 : table_remove(data, table);
2638 23 : table_free(table);
2639 23 : table = NULL;
2640 :
2641 : // init again cloudsync for the table
2642 23 : table_algo algo_current = dbutils_table_settings_get_algo(data, table_name);
2643 23 : if (algo_current == table_algo_none) algo_current = dbutils_table_settings_get_algo(data, "*");
2644 23 : rc = cloudsync_init_table(data, table_name, cloudsync_algo_name(algo_current), CLOUDSYNC_INIT_FLAG_SKIP_INT_PK_CHECK);
2645 23 : if (rc != DBRES_OK) goto rollback_finalize_alter;
2646 :
2647 23 : return DBRES_OK;
2648 :
2649 : rollback_finalize_alter:
2650 1 : if (table) {
2651 0 : table_set_pknames(table, NULL);
2652 0 : table->is_altering = false;
2653 0 : }
2654 1 : return rc;
2655 24 : }
2656 :
2657 : // MARK: - Filter Rewrite -
2658 :
2659 : // Replace bare column names in a filter expression with prefix-qualified names.
2660 : // E.g., filter="user_id = 42", prefix="NEW", columns=["user_id","id"] → "NEW.\"user_id\" = 42"
2661 : // Columns must be sorted by length descending by the caller to avoid partial matches.
2662 : // Skips content inside single-quoted string literals.
2663 : // Returns a newly allocated string (caller must free with cloudsync_memory_free), or NULL on error.
2664 : // Helper: check if an identifier token matches a column name.
2665 112 : static bool filter_is_column (const char *token, size_t token_len, char **columns, int ncols) {
2666 450 : for (int i = 0; i < ncols; ++i) {
2667 388 : if (strlen(columns[i]) == token_len && strncmp(token, columns[i], token_len) == 0)
2668 50 : return true;
2669 338 : }
2670 62 : return false;
2671 112 : }
2672 :
2673 : // Helper: check if character is part of a SQL identifier.
2674 796 : static bool filter_is_ident_char (char c) {
2675 1262 : return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
2676 466 : (c >= '0' && c <= '9') || c == '_';
2677 : }
2678 :
2679 40 : char *cloudsync_filter_add_row_prefix (const char *filter, const char *prefix, char **columns, int ncols) {
2680 40 : if (!filter || !prefix || !columns || ncols <= 0) return NULL;
2681 :
2682 40 : size_t filter_len = strlen(filter);
2683 40 : size_t prefix_len = strlen(prefix);
2684 :
2685 : // Each identifier match grows by at most (prefix_len + 3) bytes.
2686 : // Worst case: the entire filter is one repeated column reference separated by
2687 : // single characters, so up to (filter_len / 2) matches. Use a safe upper bound.
2688 40 : size_t max_growth = (filter_len / 2 + 1) * (prefix_len + 3);
2689 40 : size_t cap = filter_len + max_growth + 64;
2690 40 : char *result = (char *)cloudsync_memory_alloc(cap);
2691 40 : if (!result) return NULL;
2692 40 : size_t out = 0;
2693 :
2694 : // Single pass: tokenize into identifiers, quoted strings, and everything else.
2695 40 : size_t i = 0;
2696 336 : while (i < filter_len) {
2697 : // Skip single-quoted string literals verbatim (handle '' escape)
2698 296 : if (filter[i] == '\'') {
2699 6 : result[out++] = filter[i++];
2700 38 : while (i < filter_len) {
2701 38 : if (filter[i] == '\'') {
2702 6 : result[out++] = filter[i++];
2703 : // '' is an escaped quote — keep going
2704 6 : if (i < filter_len && filter[i] == '\'') {
2705 0 : result[out++] = filter[i++];
2706 0 : continue;
2707 : }
2708 6 : break; // single ' ends the literal
2709 : }
2710 32 : result[out++] = filter[i++];
2711 : }
2712 6 : continue;
2713 : }
2714 :
2715 : // Extract identifier token
2716 290 : if (filter_is_ident_char(filter[i])) {
2717 112 : size_t start = i;
2718 540 : while (i < filter_len && filter_is_ident_char(filter[i])) ++i;
2719 112 : size_t token_len = i - start;
2720 :
2721 112 : if (filter_is_column(&filter[start], token_len, columns, ncols)) {
2722 : // Emit PREFIX."column_name"
2723 50 : memcpy(&result[out], prefix, prefix_len); out += prefix_len;
2724 50 : result[out++] = '.';
2725 50 : result[out++] = '"';
2726 50 : memcpy(&result[out], &filter[start], token_len); out += token_len;
2727 50 : result[out++] = '"';
2728 50 : } else {
2729 : // Not a column — copy as-is
2730 62 : memcpy(&result[out], &filter[start], token_len); out += token_len;
2731 : }
2732 112 : continue;
2733 : }
2734 :
2735 : // Any other character — copy as-is
2736 178 : result[out++] = filter[i++];
2737 : }
2738 :
2739 40 : result[out] = '\0';
2740 40 : return result;
2741 40 : }
2742 :
2743 24 : int cloudsync_reset_metatable (cloudsync_context *data, const char *table_name) {
2744 24 : cloudsync_table_context *table = table_lookup(data, table_name);
2745 24 : if (!table) return DBRES_ERROR;
2746 :
2747 24 : char *sql = cloudsync_memory_mprintf(SQL_DELETE_ALL_FROM_CLOUDSYNC_TABLE, table->meta_ref);
2748 24 : int rc = database_exec(data, sql);
2749 24 : cloudsync_memory_free(sql);
2750 24 : if (rc != DBRES_OK) return rc;
2751 :
2752 24 : return cloudsync_refill_metatable(data, table_name);
2753 24 : }
2754 :
2755 326 : int cloudsync_refill_metatable (cloudsync_context *data, const char *table_name) {
2756 326 : cloudsync_table_context *table = table_lookup(data, table_name);
2757 326 : if (!table) return DBRES_ERROR;
2758 :
2759 326 : dbvm_t *vm = NULL;
2760 326 : int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
2761 :
2762 : // Read row-level filter from settings (if any)
2763 : char filter_buf[2048];
2764 326 : int frc = dbutils_table_settings_get_value(data, table_name, "*", "filter", filter_buf, sizeof(filter_buf));
2765 326 : const char *filter = (frc == DBRES_OK && filter_buf[0]) ? filter_buf : NULL;
2766 :
2767 326 : const char *schema = table->schema ? table->schema : "";
2768 326 : char *sql = sql_build_pk_collist_query(schema, table_name);
2769 326 : char *pkclause_identifiers = NULL;
2770 326 : int rc = database_select_text(data, sql, &pkclause_identifiers);
2771 326 : cloudsync_memory_free(sql);
2772 326 : if (rc != DBRES_OK) goto finalize;
2773 326 : char *pkvalues_identifiers = (pkclause_identifiers) ? pkclause_identifiers : "rowid";
2774 :
2775 : // Use database-specific query builder to handle type differences in composite PKs
2776 326 : sql = sql_build_insert_missing_pks_query(schema, table_name, pkvalues_identifiers, table->base_ref, table->meta_ref, filter);
2777 326 : if (!sql) {rc = DBRES_NOMEM; goto finalize;}
2778 326 : rc = database_exec(data, sql);
2779 326 : cloudsync_memory_free(sql);
2780 326 : if (rc != DBRES_OK) goto finalize;
2781 :
2782 : // fill missing colums
2783 : // for each non-pk column:
2784 : // The new query does 1 encode per source row and one indexed NOT-EXISTS probe.
2785 : // The old plan does many decodes per candidate and can't use an index to rule out matches quickly—so it burns CPU and I/O.
2786 :
2787 326 : if (filter) {
2788 20 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_PKS_NOT_IN_SYNC_FOR_COL_FILTERED, pkvalues_identifiers, table->base_ref, filter, table->meta_ref);
2789 20 : } else {
2790 306 : sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_SELECT_PKS_NOT_IN_SYNC_FOR_COL, pkvalues_identifiers, table->base_ref, table->meta_ref);
2791 : }
2792 326 : rc = databasevm_prepare(data, sql, (void **)&vm, DBFLAG_PERSISTENT);
2793 326 : cloudsync_memory_free(sql);
2794 326 : if (rc != DBRES_OK) goto finalize;
2795 :
2796 1508 : for (int i=0; i<table->ncols; ++i) {
2797 1182 : char *col_name = table->col_name[i];
2798 :
2799 1182 : rc = databasevm_bind_text(vm, 1, col_name, -1);
2800 1182 : if (rc != DBRES_OK) goto finalize;
2801 :
2802 1220 : while (1) {
2803 1220 : rc = databasevm_step(vm);
2804 1220 : if (rc == DBRES_ROW) {
2805 38 : size_t pklen = 0;
2806 38 : const void *pk = (const char *)database_column_blob(vm, 0, &pklen);
2807 38 : if (!pk) { rc = DBRES_ERROR; break; }
2808 38 : rc = local_mark_insert_or_update_meta(table, pk, pklen, col_name, db_version, cloudsync_bumpseq(data));
2809 1220 : } else if (rc == DBRES_DONE) {
2810 1182 : rc = DBRES_OK;
2811 1182 : break;
2812 : } else {
2813 0 : break;
2814 : }
2815 : }
2816 1182 : if (rc != DBRES_OK) goto finalize;
2817 :
2818 1182 : databasevm_reset(vm);
2819 1508 : }
2820 :
2821 : finalize:
2822 326 : if (rc != DBRES_OK) {DEBUG_ALWAYS("cloudsync_refill_metatable error: %s", database_errmsg(data));}
2823 326 : if (pkclause_identifiers) cloudsync_memory_free(pkclause_identifiers);
2824 326 : if (vm) databasevm_finalize(vm);
2825 326 : return rc;
2826 326 : }
2827 :
2828 : // MARK: - Local -
2829 :
2830 4 : int local_update_sentinel (cloudsync_table_context *table, const void *pk, size_t pklen, int64_t db_version, int seq) {
2831 4 : dbvm_t *vm = table->meta_sentinel_update_stmt;
2832 4 : if (!vm) return -1;
2833 :
2834 4 : int rc = databasevm_bind_int(vm, 1, db_version);
2835 4 : if (rc != DBRES_OK) goto cleanup;
2836 :
2837 4 : rc = databasevm_bind_int(vm, 2, seq);
2838 4 : if (rc != DBRES_OK) goto cleanup;
2839 :
2840 4 : rc = databasevm_bind_blob(vm, 3, pk, (int)pklen);
2841 4 : if (rc != DBRES_OK) goto cleanup;
2842 :
2843 4 : rc = databasevm_step(vm);
2844 4 : if (rc == DBRES_DONE) rc = DBRES_OK;
2845 :
2846 : cleanup:
2847 4 : DEBUG_DBERROR(rc, "local_update_sentinel", table->context);
2848 4 : databasevm_reset(vm);
2849 4 : return rc;
2850 4 : }
2851 :
2852 126 : int local_mark_insert_sentinel_meta (cloudsync_table_context *table, const void *pk, size_t pklen, int64_t db_version, int seq) {
2853 126 : dbvm_t *vm = table->meta_sentinel_insert_stmt;
2854 126 : if (!vm) return -1;
2855 :
2856 126 : int rc = databasevm_bind_blob(vm, 1, pk, (int)pklen);
2857 126 : if (rc != DBRES_OK) goto cleanup;
2858 :
2859 126 : rc = databasevm_bind_int(vm, 2, db_version);
2860 126 : if (rc != DBRES_OK) goto cleanup;
2861 :
2862 126 : rc = databasevm_bind_int(vm, 3, seq);
2863 126 : if (rc != DBRES_OK) goto cleanup;
2864 :
2865 126 : rc = databasevm_bind_int(vm, 4, db_version);
2866 126 : if (rc != DBRES_OK) goto cleanup;
2867 :
2868 126 : rc = databasevm_bind_int(vm, 5, seq);
2869 126 : if (rc != DBRES_OK) goto cleanup;
2870 :
2871 126 : rc = databasevm_step(vm);
2872 126 : if (rc == DBRES_DONE) rc = DBRES_OK;
2873 :
2874 : cleanup:
2875 126 : DEBUG_DBERROR(rc, "local_insert_sentinel", table->context);
2876 126 : databasevm_reset(vm);
2877 126 : return rc;
2878 126 : }
2879 :
2880 13493 : int local_mark_insert_or_update_meta_impl (cloudsync_table_context *table, const void *pk, size_t pklen, const char *col_name, int col_version, int64_t db_version, int seq) {
2881 :
2882 13493 : dbvm_t *vm = table->meta_row_insert_update_stmt;
2883 13493 : if (!vm) return -1;
2884 :
2885 13493 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
2886 13493 : if (rc != DBRES_OK) goto cleanup;
2887 :
2888 13493 : rc = databasevm_bind_text(vm, 2, (col_name) ? col_name : CLOUDSYNC_TOMBSTONE_VALUE, -1);
2889 13493 : if (rc != DBRES_OK) goto cleanup;
2890 :
2891 13493 : rc = databasevm_bind_int(vm, 3, col_version);
2892 13493 : if (rc != DBRES_OK) goto cleanup;
2893 :
2894 13493 : rc = databasevm_bind_int(vm, 4, db_version);
2895 13493 : if (rc != DBRES_OK) goto cleanup;
2896 :
2897 13493 : rc = databasevm_bind_int(vm, 5, seq);
2898 13493 : if (rc != DBRES_OK) goto cleanup;
2899 :
2900 13493 : rc = databasevm_bind_int(vm, 6, db_version);
2901 13493 : if (rc != DBRES_OK) goto cleanup;
2902 :
2903 13493 : rc = databasevm_bind_int(vm, 7, seq);
2904 13493 : if (rc != DBRES_OK) goto cleanup;
2905 :
2906 13493 : rc = databasevm_step(vm);
2907 13493 : if (rc == DBRES_DONE) rc = DBRES_OK;
2908 :
2909 : cleanup:
2910 13493 : DEBUG_DBERROR(rc, "local_insert_or_update", table->context);
2911 13493 : databasevm_reset(vm);
2912 13493 : return rc;
2913 13493 : }
2914 :
2915 13387 : int local_mark_insert_or_update_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const char *col_name, int64_t db_version, int seq) {
2916 13387 : return local_mark_insert_or_update_meta_impl(table, pk, pklen, col_name, 1, db_version, seq);
2917 : }
2918 :
2919 41 : int local_mark_delete_block_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const char *block_colname, int64_t db_version, int seq) {
2920 : // Mark a block as deleted by setting col_version = 2 (even = deleted)
2921 41 : return local_mark_insert_or_update_meta_impl(table, pk, pklen, block_colname, 2, db_version, seq);
2922 : }
2923 :
2924 41 : int block_delete_value_external (cloudsync_context *data, cloudsync_table_context *table, const void *pk, size_t pklen, const char *block_colname) {
2925 41 : return block_delete_value(data, table, pk, (int)pklen, block_colname);
2926 : }
2927 :
2928 65 : int local_mark_delete_meta (cloudsync_table_context *table, const void *pk, size_t pklen, int64_t db_version, int seq) {
2929 65 : return local_mark_insert_or_update_meta_impl(table, pk, pklen, NULL, 2, db_version, seq);
2930 : }
2931 :
2932 36 : int local_drop_meta (cloudsync_table_context *table, const void *pk, size_t pklen) {
2933 36 : dbvm_t *vm = table->meta_row_drop_stmt;
2934 36 : if (!vm) return -1;
2935 :
2936 36 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
2937 36 : if (rc != DBRES_OK) goto cleanup;
2938 :
2939 36 : rc = databasevm_step(vm);
2940 36 : if (rc == DBRES_DONE) rc = DBRES_OK;
2941 :
2942 : cleanup:
2943 36 : DEBUG_DBERROR(rc, "local_drop_meta", table->context);
2944 36 : databasevm_reset(vm);
2945 36 : return rc;
2946 36 : }
2947 :
2948 29 : int local_update_move_meta (cloudsync_table_context *table, const void *pk, size_t pklen, const void *pk2, size_t pklen2, int64_t db_version) {
2949 : /*
2950 : * This function moves non-sentinel metadata entries from an old primary key (OLD.pk)
2951 : * to a new primary key (NEW.pk) when a primary key change occurs.
2952 : *
2953 : * To ensure consistency and proper conflict resolution in a CRDT (Conflict-free Replicated Data Type) system,
2954 : * each non-sentinel metadata entry involved in the move must have a unique sequence value (seq).
2955 : *
2956 : * The `seq` is crucial for tracking the order of operations and for detecting and resolving conflicts
2957 : * during synchronization between replicas. Without a unique `seq` for each entry, concurrent updates
2958 : * may be applied incorrectly, leading to data inconsistency.
2959 : *
2960 : * When performing the update, a unique `seq` must be assigned to each metadata row. This can be achieved
2961 : * by either incrementing the maximum sequence value in the table or using a function (e.g., cloudsync_bumpseq(data))
2962 : * that generates a unique sequence for each row. The update query should ensure that each row moved
2963 : * from OLD.pk to NEW.pk gets a distinct `seq` to maintain proper versioning and ordering of changes.
2964 : */
2965 :
2966 : // see https://github.com/sqliteai/sqlite-sync/blob/main/docs/PriKey.md for more details
2967 : // pk2 is the old pk
2968 :
2969 29 : dbvm_t *vm = table->meta_update_move_stmt;
2970 29 : if (!vm) return -1;
2971 :
2972 : // new primary key
2973 29 : int rc = databasevm_bind_blob(vm, 1, pk, pklen);
2974 29 : if (rc != DBRES_OK) goto cleanup;
2975 :
2976 : // new db_version
2977 29 : rc = databasevm_bind_int(vm, 2, db_version);
2978 29 : if (rc != DBRES_OK) goto cleanup;
2979 :
2980 : // old primary key
2981 29 : rc = databasevm_bind_blob(vm, 3, pk2, pklen2);
2982 29 : if (rc != DBRES_OK) goto cleanup;
2983 :
2984 29 : rc = databasevm_step(vm);
2985 29 : if (rc == DBRES_DONE) rc = DBRES_OK;
2986 :
2987 : cleanup:
2988 29 : DEBUG_DBERROR(rc, "local_update_move_meta", table->context);
2989 29 : databasevm_reset(vm);
2990 29 : return rc;
2991 29 : }
2992 :
2993 : // MARK: - Payload Encode / Decode -
2994 :
2995 743 : static void cloudsync_payload_checksum_store (cloudsync_payload_header *header, uint64_t checksum) {
2996 743 : uint64_t h = checksum & 0xFFFFFFFFFFFFULL; // keep 48 bits
2997 743 : header->checksum[0] = (uint8_t)(h >> 40);
2998 743 : header->checksum[1] = (uint8_t)(h >> 32);
2999 743 : header->checksum[2] = (uint8_t)(h >> 24);
3000 743 : header->checksum[3] = (uint8_t)(h >> 16);
3001 743 : header->checksum[4] = (uint8_t)(h >> 8);
3002 743 : header->checksum[5] = (uint8_t)(h >> 0);
3003 743 : }
3004 :
3005 748 : static uint64_t cloudsync_payload_checksum_load (cloudsync_payload_header *header) {
3006 2244 : return ((uint64_t)header->checksum[0] << 40) |
3007 1496 : ((uint64_t)header->checksum[1] << 32) |
3008 1496 : ((uint64_t)header->checksum[2] << 24) |
3009 1496 : ((uint64_t)header->checksum[3] << 16) |
3010 1496 : ((uint64_t)header->checksum[4] << 8) |
3011 748 : ((uint64_t)header->checksum[5] << 0);
3012 : }
3013 :
3014 748 : static bool cloudsync_payload_checksum_verify (cloudsync_payload_header *header, uint64_t checksum) {
3015 748 : uint64_t checksum1 = cloudsync_payload_checksum_load(header);
3016 748 : uint64_t checksum2 = checksum & 0xFFFFFFFFFFFFULL;
3017 748 : return (checksum1 == checksum2);
3018 : }
3019 :
3020 52295 : static bool cloudsync_payload_encode_check (cloudsync_payload_context *payload, size_t needed) {
3021 52295 : if (payload->nrows == 0) needed += sizeof(cloudsync_payload_header);
3022 :
3023 : // alloc/resize buffer
3024 52295 : if (payload->bused + needed > payload->balloc) {
3025 762 : if (needed < CLOUDSYNC_PAYLOAD_MINBUF_SIZE) needed = CLOUDSYNC_PAYLOAD_MINBUF_SIZE;
3026 762 : size_t balloc = payload->balloc + needed;
3027 :
3028 762 : char *buffer = cloudsync_memory_realloc(payload->buffer, balloc);
3029 762 : if (!buffer) {
3030 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3031 0 : memset(payload, 0, sizeof(cloudsync_payload_context));
3032 0 : return false;
3033 : }
3034 :
3035 762 : payload->buffer = buffer;
3036 762 : payload->balloc = balloc;
3037 762 : if (payload->nrows == 0) payload->bused = sizeof(cloudsync_payload_header);
3038 762 : }
3039 :
3040 52295 : return true;
3041 52295 : }
3042 :
3043 51241 : size_t cloudsync_payload_context_size (size_t *header_size) {
3044 51241 : if (header_size) *header_size = sizeof(cloudsync_payload_header);
3045 51241 : return sizeof(cloudsync_payload_context);
3046 : }
3047 :
3048 0 : void cloudsync_payload_context_free (cloudsync_payload_context *payload) {
3049 0 : if (!payload) return;
3050 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3051 0 : cloudsync_memory_free(payload);
3052 0 : }
3053 :
3054 4108 : uint64_t cloudsync_payload_context_nrows (cloudsync_payload_context *payload) {
3055 4108 : return payload ? payload->nrows : 0;
3056 : }
3057 :
3058 2021 : size_t cloudsync_payload_context_bused (cloudsync_payload_context *payload) {
3059 2021 : return payload ? payload->bused : 0;
3060 : }
3061 :
3062 743 : void cloudsync_payload_header_init (cloudsync_payload_header *header, uint8_t version, uint32_t expanded_size, uint16_t ncols, uint32_t nrows, uint64_t hash) {
3063 743 : memset(header, 0, sizeof(cloudsync_payload_header));
3064 : assert(sizeof(cloudsync_payload_header)==32);
3065 :
3066 : int major, minor, patch;
3067 743 : sscanf(CLOUDSYNC_VERSION, "%d.%d.%d", &major, &minor, &patch);
3068 :
3069 743 : header->signature = htonl(CLOUDSYNC_PAYLOAD_SIGNATURE);
3070 743 : header->version = version;
3071 743 : header->libversion[0] = (uint8_t)major;
3072 743 : header->libversion[1] = (uint8_t)minor;
3073 743 : header->libversion[2] = (uint8_t)patch;
3074 743 : header->expanded_size = htonl(expanded_size);
3075 743 : header->ncols = htons(ncols);
3076 743 : header->nrows = htonl(nrows);
3077 743 : header->schema_hash = htonll(hash);
3078 743 : }
3079 :
3080 52263 : int cloudsync_payload_encode_step (cloudsync_payload_context *payload, cloudsync_context *data, int argc, dbvalue_t **argv) {
3081 : DEBUG_FUNCTION("cloudsync_payload_encode_step");
3082 : // debug_values(argc, argv);
3083 :
3084 : // check if the step function is called for the first time
3085 52263 : if (payload->nrows == 0) payload->ncols = (uint16_t)argc;
3086 :
3087 52263 : size_t breq = pk_encode_size((dbvalue_t **)argv, argc, 0, data->skip_decode_idx);
3088 52263 : if (cloudsync_payload_encode_check(payload, breq) == false) {
3089 0 : return cloudsync_set_error(data, "Not enough memory to resize payload internal buffer", DBRES_NOMEM);
3090 : }
3091 :
3092 52263 : char *buffer = payload->buffer + payload->bused;
3093 52263 : size_t bsize = payload->balloc - payload->bused;
3094 52263 : char *p = pk_encode((dbvalue_t **)argv, argc, buffer, false, &bsize, data->skip_decode_idx);
3095 52263 : if (!p) return cloudsync_set_error(data, "An error occurred while encoding payload", DBRES_ERROR);
3096 :
3097 : // update buffer
3098 52263 : payload->bused += breq;
3099 :
3100 : // increment row counter
3101 52263 : ++payload->nrows;
3102 :
3103 52263 : return DBRES_OK;
3104 52263 : }
3105 :
3106 32 : static bool cloudsync_payload_append_raw (cloudsync_payload_context *payload, cloudsync_context *data, const char **fields, const size_t *field_sizes, int nfields, uint8_t version) {
3107 32 : size_t needed = 0;
3108 320 : for (int i = 0; i < nfields; ++i) {
3109 288 : if (field_sizes[i] > SIZE_MAX - needed) {
3110 0 : cloudsync_set_error(data, CLOUDSYNC_ERRCODE_ROW_TOO_LARGE "cloudsync payload raw row too large", DBRES_NOMEM);
3111 0 : return false;
3112 : }
3113 288 : needed += field_sizes[i];
3114 288 : }
3115 32 : if (!cloudsync_payload_encode_check(payload, needed)) {
3116 0 : cloudsync_set_error(data, "Not enough memory to resize payload internal buffer", DBRES_NOMEM);
3117 0 : return false;
3118 : }
3119 32 : if (payload->nrows == 0) {
3120 32 : payload->ncols = (uint16_t)nfields;
3121 32 : payload->version = version;
3122 32 : }
3123 32 : char *dst = payload->buffer + payload->bused;
3124 320 : for (int i = 0; i < nfields; ++i) {
3125 288 : memcpy(dst, fields[i], field_sizes[i]);
3126 288 : dst += field_sizes[i];
3127 288 : }
3128 32 : payload->bused += needed;
3129 32 : ++payload->nrows;
3130 32 : return true;
3131 32 : }
3132 :
3133 56 : int cloudsync_payload_max_chunk_size (cloudsync_context *data) {
3134 56 : int64_t value = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_PAYLOAD_MAX_CHUNK_SIZE);
3135 56 : if (value <= 0) value = CLOUDSYNC_PAYLOAD_CHUNK_DEFAULT_SIZE;
3136 56 : if (value < CLOUDSYNC_PAYLOAD_CHUNK_MIN_SIZE) value = CLOUDSYNC_PAYLOAD_CHUNK_MIN_SIZE;
3137 56 : if (value > CLOUDSYNC_PAYLOAD_CHUNK_MAX_SIZE) value = CLOUDSYNC_PAYLOAD_CHUNK_MAX_SIZE;
3138 56 : return (int)value;
3139 : }
3140 :
3141 0 : int cloudsync_payload_fragment_target_size (cloudsync_context *data) {
3142 0 : int max_size = cloudsync_payload_max_chunk_size(data);
3143 0 : int target = max_size - (int)sizeof(cloudsync_payload_header) - CLOUDSYNC_PAYLOAD_CHUNK_SAFETY_MARGIN;
3144 0 : if (target < 1024) target = 1024;
3145 0 : return target;
3146 : }
3147 :
3148 84 : static size_t cloudsync_payload_decimal_len_i64 (int64_t value) {
3149 84 : size_t len = value < 0 ? 1 : 0;
3150 84 : uint64_t v = (value < 0) ? (uint64_t)(-(value + 1)) + 1u : (uint64_t)value;
3151 84 : do {
3152 240 : len++;
3153 240 : v /= 10u;
3154 240 : } while (v != 0);
3155 84 : return len;
3156 : }
3157 :
3158 224 : static bool cloudsync_payload_size_add (size_t *acc, size_t value) {
3159 224 : if (value > SIZE_MAX - *acc) return false;
3160 224 : *acc += value;
3161 224 : return true;
3162 224 : }
3163 :
3164 28 : int cloudsync_payload_fragment_count (int64_t total_size, int target_size) {
3165 28 : if (total_size <= 0 || target_size <= 0) return 0;
3166 28 : uint64_t total = (uint64_t)total_size;
3167 28 : uint64_t target = (uint64_t)target_size;
3168 28 : uint64_t count = total / target + ((total % target) != 0);
3169 28 : if (count == 0 || count > INT_MAX) return 0;
3170 28 : return (int)count;
3171 28 : }
3172 :
3173 28 : int cloudsync_payload_fragment_data_size (cloudsync_context *data,
3174 : const char *tbl, int tbl_len,
3175 : const void *pk, int pk_len,
3176 : const char *col_name, int col_name_len,
3177 : int64_t col_version, int64_t db_version,
3178 : const void *site_id, int site_id_len,
3179 : int64_t cl, int64_t seq,
3180 : int64_t total_size,
3181 : int part_index, int part_count) {
3182 28 : UNUSED_PARAMETER(pk);
3183 28 : UNUSED_PARAMETER(site_id);
3184 28 : if (tbl_len < 0 && tbl) tbl_len = (int)strlen(tbl);
3185 28 : if (col_name_len < 0 && col_name) col_name_len = (int)strlen(col_name);
3186 28 : if (tbl_len < 0 || pk_len < 0 || col_name_len < 0 || site_id_len < 0 || total_size < 0 || part_index < 0 || part_count <= 0) {
3187 0 : return 0;
3188 : }
3189 :
3190 28 : size_t fixed = sizeof(cloudsync_payload_header);
3191 56 : size_t frag_col_len = strlen(CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX) + 32 + 1 + 16 + 1 +
3192 84 : cloudsync_payload_decimal_len_i64(part_index) + 1 +
3193 84 : cloudsync_payload_decimal_len_i64(part_count) + 1 +
3194 84 : cloudsync_payload_decimal_len_i64(total_size) + 1 +
3195 28 : (size_t)col_name_len;
3196 224 : size_t sizes[] = {
3197 28 : pk_encode_raw_size(DBTYPE_TEXT, tbl_len),
3198 28 : pk_encode_raw_size(DBTYPE_BLOB, pk_len),
3199 28 : pk_encode_raw_size(DBTYPE_TEXT, (int64_t)frag_col_len),
3200 28 : pk_encode_raw_size(DBTYPE_INTEGER, col_version),
3201 28 : pk_encode_raw_size(DBTYPE_INTEGER, db_version),
3202 28 : pk_encode_raw_size(DBTYPE_BLOB, site_id_len),
3203 28 : pk_encode_raw_size(DBTYPE_INTEGER, cl),
3204 28 : pk_encode_raw_size(DBTYPE_INTEGER, seq)
3205 : };
3206 252 : for (size_t i = 0; i < sizeof(sizes) / sizeof(sizes[0]); ++i) {
3207 224 : if (sizes[i] == SIZE_MAX || !cloudsync_payload_size_add(&fixed, sizes[i])) return 0;
3208 224 : }
3209 :
3210 28 : int max_size = cloudsync_payload_max_chunk_size(data);
3211 28 : if (fixed >= (size_t)max_size) return 0;
3212 :
3213 28 : size_t candidate = (size_t)max_size - fixed;
3214 28 : if (candidate > INT_MAX) candidate = INT_MAX;
3215 56 : while (candidate > 0) {
3216 56 : size_t frag_size = pk_encode_raw_size(DBTYPE_BLOB, (int64_t)candidate);
3217 56 : if (frag_size == SIZE_MAX) return 0;
3218 56 : if (fixed <= (size_t)max_size && frag_size <= (size_t)max_size - fixed) return (int)candidate;
3219 28 : size_t total = fixed + frag_size;
3220 28 : size_t over = total > (size_t)max_size ? total - (size_t)max_size : 1;
3221 28 : if (candidate <= over) return 0;
3222 28 : candidate -= over;
3223 : }
3224 0 : return 0;
3225 28 : }
3226 :
3227 28 : int cloudsync_payload_encoded_value_header (dbvalue_t *value, char *header, int header_cap, int64_t *payload_len) {
3228 28 : if (!value || !header || header_cap <= 0 || !payload_len) return -1;
3229 28 : int type = database_value_type(value);
3230 28 : *payload_len = 0;
3231 28 : if (type != DBTYPE_TEXT && type != DBTYPE_BLOB) return 0;
3232 28 : int64_t len = database_value_bytes(value);
3233 28 : if (len < 0) return -1;
3234 28 : *payload_len = len;
3235 28 : size_t total = pk_encode_raw_size(type, len);
3236 28 : if (total == SIZE_MAX || total < (size_t)len || total - (size_t)len > (size_t)header_cap) return -1;
3237 28 : if (type == DBTYPE_TEXT) {
3238 18 : size_t nbytes = pk_encode_raw_size(type, len) - (size_t)len - 1;
3239 18 : uint8_t type_byte = (uint8_t)((nbytes << 3) | DBTYPE_TEXT);
3240 18 : header[0] = (char)type_byte;
3241 72 : for (size_t i = 0; i < nbytes; i++) header[1 + i] = (uint8_t)(((uint64_t)len >> (8 * (nbytes - 1 - i))) & 0xFFu);
3242 18 : return (int)(1 + nbytes);
3243 : } else {
3244 10 : size_t nbytes = pk_encode_raw_size(type, len) - (size_t)len - 1;
3245 10 : uint8_t type_byte = (uint8_t)((nbytes << 3) | DBTYPE_BLOB);
3246 10 : header[0] = (char)type_byte;
3247 40 : for (size_t i = 0; i < nbytes; i++) header[1 + i] = (uint8_t)(((uint64_t)len >> (8 * (nbytes - 1 - i))) & 0xFFu);
3248 10 : return (int)(1 + nbytes);
3249 : }
3250 28 : }
3251 :
3252 14 : uint64_t cloudsync_payload_encoded_value_checksum (dbvalue_t *value) {
3253 14 : if (!value) return 0;
3254 14 : int type = database_value_type(value);
3255 14 : if (type != DBTYPE_TEXT && type != DBTYPE_BLOB) {
3256 0 : size_t len = pk_encode_size(&value, 1, 0, -1);
3257 : char stack[32];
3258 0 : char *buf = stack;
3259 0 : if (len > sizeof(stack)) buf = cloudsync_memory_alloc((uint64_t)len);
3260 0 : if (!buf) return 0;
3261 0 : size_t bsize = len;
3262 0 : pk_encode(&value, 1, buf, false, &bsize, -1);
3263 0 : uint64_t h = pk_checksum(buf, bsize);
3264 0 : if (buf != stack) cloudsync_memory_free(buf);
3265 0 : return h;
3266 : }
3267 : char header[16];
3268 14 : int64_t payload_len = 0;
3269 14 : int header_len = cloudsync_payload_encoded_value_header(value, header, sizeof(header), &payload_len);
3270 14 : if (header_len <= 0) return 0;
3271 14 : uint64_t h = pk_checksum(header, (size_t)header_len);
3272 14 : const char *p = (const char *)database_value_blob(value);
3273 14 : if (p && payload_len > 0) {
3274 14 : const uint8_t *bytes = (const uint8_t *)p;
3275 18720014 : for (int64_t i = 0; i < payload_len; ++i) {
3276 18720000 : h ^= bytes[i];
3277 18720000 : h *= 1099511628211ULL;
3278 18720000 : }
3279 14 : }
3280 14 : return h;
3281 14 : }
3282 :
3283 7634 : static uint64_t cloudsync_checksum_update (uint64_t h, const void *data, size_t len) {
3284 7634 : const uint8_t *p = (const uint8_t *)data;
3285 9020067 : for (size_t i = 0; i < len; ++i) {
3286 9012433 : h ^= p[i];
3287 9012433 : h *= 1099511628211ULL;
3288 9012433 : }
3289 7634 : return h;
3290 : }
3291 :
3292 852 : static uint64_t cloudsync_checksum_update_i64 (uint64_t h, int64_t value) {
3293 852 : uint64_t v = (uint64_t)value;
3294 7668 : for (int i = 7; i >= 0; --i) {
3295 6816 : uint8_t b = (uint8_t)((v >> (8 * i)) & 0xffu);
3296 6816 : h = cloudsync_checksum_update(h, &b, 1);
3297 6816 : }
3298 852 : return h;
3299 : }
3300 :
3301 71 : static void cloudsync_payload_fragment_value_id (char out[33],
3302 : const char *tbl, int tbl_len,
3303 : const void *pk, int pk_len,
3304 : const char *col_name, int col_name_len,
3305 : int64_t col_version, int64_t db_version,
3306 : const void *site_id, int site_id_len,
3307 : int64_t cl, int64_t seq,
3308 : uint64_t value_checksum,
3309 : int64_t total_size) {
3310 71 : uint64_t h1 = 14695981039346656037ULL;
3311 71 : uint64_t h2 = 1099511628211ULL;
3312 71 : const char sep = '\x1f';
3313 :
3314 71 : h1 = cloudsync_checksum_update(h1, tbl, (size_t)tbl_len);
3315 71 : h1 = cloudsync_checksum_update(h1, &sep, 1);
3316 71 : h1 = cloudsync_checksum_update(h1, pk, (size_t)pk_len);
3317 71 : h1 = cloudsync_checksum_update(h1, &sep, 1);
3318 71 : h1 = cloudsync_checksum_update(h1, col_name, (size_t)col_name_len);
3319 71 : h1 = cloudsync_checksum_update(h1, &sep, 1);
3320 71 : h1 = cloudsync_checksum_update(h1, site_id, (size_t)site_id_len);
3321 71 : h1 = cloudsync_checksum_update_i64(h1, col_version);
3322 71 : h1 = cloudsync_checksum_update_i64(h1, db_version);
3323 71 : h1 = cloudsync_checksum_update_i64(h1, cl);
3324 71 : h1 = cloudsync_checksum_update_i64(h1, seq);
3325 71 : h1 = cloudsync_checksum_update_i64(h1, (int64_t)value_checksum);
3326 71 : h1 = cloudsync_checksum_update_i64(h1, total_size);
3327 :
3328 71 : h2 = cloudsync_checksum_update_i64(h2, total_size);
3329 71 : h2 = cloudsync_checksum_update_i64(h2, (int64_t)value_checksum);
3330 71 : h2 = cloudsync_checksum_update(h2, site_id, (size_t)site_id_len);
3331 71 : h2 = cloudsync_checksum_update(h2, col_name, (size_t)col_name_len);
3332 71 : h2 = cloudsync_checksum_update(h2, pk, (size_t)pk_len);
3333 71 : h2 = cloudsync_checksum_update(h2, tbl, (size_t)tbl_len);
3334 71 : h2 = cloudsync_checksum_update_i64(h2, seq);
3335 71 : h2 = cloudsync_checksum_update_i64(h2, cl);
3336 71 : h2 = cloudsync_checksum_update_i64(h2, db_version);
3337 71 : h2 = cloudsync_checksum_update_i64(h2, col_version);
3338 :
3339 71 : snprintf(out, 33, "%016" PRIx64 "%016" PRIx64, h1, h2);
3340 71 : }
3341 :
3342 32 : int cloudsync_payload_encode_fragment_step (cloudsync_payload_context *payload, cloudsync_context *data,
3343 : const char *tbl, int tbl_len,
3344 : const void *pk, int pk_len,
3345 : const char *col_name, int col_name_len,
3346 : const void *fragment, int fragment_len,
3347 : int64_t col_version, int64_t db_version,
3348 : const void *site_id, int site_id_len,
3349 : int64_t cl, int64_t seq,
3350 : uint64_t value_checksum,
3351 : int64_t total_size,
3352 : int part_index, int part_count) {
3353 32 : if (!payload || !data || !tbl || !pk || !col_name || !fragment || !site_id) return DBRES_MISUSE;
3354 32 : if (tbl_len < 0) tbl_len = (int)strlen(tbl);
3355 32 : if (col_name_len < 0) col_name_len = (int)strlen(col_name);
3356 64 : if (tbl_len < 0 || pk_len < 0 || col_name_len < 0 || fragment_len <= 0 || site_id_len < 0 ||
3357 32 : total_size <= 0 || part_index < 0 || part_count <= 0 || part_index >= part_count) {
3358 0 : return DBRES_MISUSE;
3359 : }
3360 :
3361 : char value_id[33];
3362 : char checksum_hex[17];
3363 64 : cloudsync_payload_fragment_value_id(value_id, tbl, tbl_len, pk, pk_len, col_name, col_name_len,
3364 32 : col_version, db_version, site_id, site_id_len, cl, seq,
3365 32 : value_checksum, total_size);
3366 32 : snprintf(checksum_hex, sizeof(checksum_hex), "%016" PRIx64, value_checksum);
3367 :
3368 32 : char *frag_col = cloudsync_memory_mprintf("%s%s:%s:%d:%d:%" PRId64 ":%.*s",
3369 : CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX,
3370 32 : value_id, checksum_hex, part_index, part_count, total_size,
3371 32 : col_name_len, col_name);
3372 32 : if (!frag_col) return DBRES_NOMEM;
3373 :
3374 32 : size_t sizes[9] = {0};
3375 32 : sizes[0] = pk_encode_raw_size(DBTYPE_TEXT, tbl_len);
3376 32 : sizes[1] = pk_encode_raw_size(DBTYPE_BLOB, pk_len);
3377 32 : sizes[2] = pk_encode_raw_size(DBTYPE_TEXT, (int64_t)strlen(frag_col));
3378 32 : sizes[3] = pk_encode_raw_size(DBTYPE_BLOB, fragment_len);
3379 32 : sizes[4] = pk_encode_raw_size(DBTYPE_INTEGER, col_version);
3380 32 : sizes[5] = pk_encode_raw_size(DBTYPE_INTEGER, db_version);
3381 32 : sizes[6] = pk_encode_raw_size(DBTYPE_BLOB, site_id_len);
3382 32 : sizes[7] = pk_encode_raw_size(DBTYPE_INTEGER, cl);
3383 32 : sizes[8] = pk_encode_raw_size(DBTYPE_INTEGER, seq);
3384 320 : for (int i = 0; i < 9; ++i) {
3385 288 : if (sizes[i] == SIZE_MAX) { cloudsync_memory_free(frag_col); return DBRES_NOMEM; }
3386 288 : }
3387 :
3388 : char stack[9][64];
3389 32 : char *fields[9] = {0};
3390 320 : for (int i = 0; i < 9; ++i) {
3391 288 : fields[i] = sizes[i] <= sizeof(stack[0]) ? stack[i] : cloudsync_memory_alloc((uint64_t)sizes[i]);
3392 288 : if (!fields[i]) {
3393 0 : for (int j = 0; j < i; ++j) if (fields[j] && (fields[j] < (char *)stack || fields[j] >= (char *)(stack + 9))) cloudsync_memory_free(fields[j]);
3394 0 : cloudsync_memory_free(frag_col);
3395 0 : return DBRES_NOMEM;
3396 : }
3397 288 : }
3398 :
3399 32 : pk_encode_raw_text(fields[0], tbl, (size_t)tbl_len);
3400 32 : pk_encode_raw_blob(fields[1], pk, (size_t)pk_len);
3401 32 : pk_encode_raw_text(fields[2], frag_col, strlen(frag_col));
3402 32 : pk_encode_raw_blob(fields[3], fragment, (size_t)fragment_len);
3403 32 : pk_encode_raw_int(fields[4], col_version);
3404 32 : pk_encode_raw_int(fields[5], db_version);
3405 32 : pk_encode_raw_blob(fields[6], site_id, (size_t)site_id_len);
3406 32 : pk_encode_raw_int(fields[7], cl);
3407 32 : pk_encode_raw_int(fields[8], seq);
3408 :
3409 : const char *cfields[9];
3410 320 : for (int i = 0; i < 9; ++i) cfields[i] = fields[i];
3411 32 : bool ok = cloudsync_payload_append_raw(payload, data, cfields, sizes, 9, CLOUDSYNC_PAYLOAD_VERSION_3);
3412 :
3413 320 : for (int i = 0; i < 9; ++i) {
3414 288 : if (!(fields[i] >= (char *)stack && fields[i] < (char *)(stack + 9))) cloudsync_memory_free(fields[i]);
3415 288 : }
3416 32 : cloudsync_memory_free(frag_col);
3417 32 : return ok ? DBRES_OK : cloudsync_errcode(data);
3418 32 : }
3419 :
3420 751 : int cloudsync_payload_encode_final (cloudsync_payload_context *payload, cloudsync_context *data) {
3421 : DEBUG_FUNCTION("cloudsync_payload_encode_final");
3422 :
3423 751 : if (payload->nrows == 0) {
3424 8 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3425 8 : payload->buffer = NULL;
3426 8 : payload->bsize = 0;
3427 8 : return DBRES_OK;
3428 : }
3429 :
3430 743 : if (payload->nrows > UINT32_MAX) {
3431 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3432 0 : payload->buffer = NULL;
3433 0 : payload->bsize = 0;
3434 0 : cloudsync_set_error(data, "Maximum number of payload rows reached", DBRES_ERROR);
3435 0 : return DBRES_ERROR;
3436 : }
3437 :
3438 : // sanity check about buffer size
3439 743 : int header_size = (int)sizeof(cloudsync_payload_header);
3440 743 : int64_t buffer_size = (int64_t)payload->bused - (int64_t)header_size;
3441 743 : if (buffer_size < 0) {
3442 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3443 0 : payload->buffer = NULL;
3444 0 : payload->bsize = 0;
3445 0 : cloudsync_set_error(data, "cloudsync_encode: internal size underflow", DBRES_ERROR);
3446 0 : return DBRES_ERROR;
3447 : }
3448 743 : if (buffer_size > INT_MAX) {
3449 0 : if (payload->buffer) cloudsync_memory_free(payload->buffer);
3450 0 : payload->buffer = NULL;
3451 0 : payload->bsize = 0;
3452 0 : cloudsync_set_error(data, CLOUDSYNC_ERRCODE_PAYLOAD_TOO_LARGE "cloudsync_encode: payload too large to compress (INT_MAX limit)", DBRES_ERROR);
3453 0 : return DBRES_ERROR;
3454 : }
3455 : // try to allocate buffer used for compressed data
3456 743 : int real_buffer_size = (int)buffer_size;
3457 743 : int zbound = LZ4_compressBound(real_buffer_size);
3458 743 : char *zbuffer = cloudsync_memory_alloc(zbound + header_size); // if for some reasons allocation fails then just skip compression
3459 :
3460 : // skip the reserved header from the buffer to compress
3461 743 : char *src_buffer = payload->buffer + sizeof(cloudsync_payload_header);
3462 743 : int zused = (zbuffer) ? LZ4_compress_default(src_buffer, zbuffer+header_size, real_buffer_size, zbound) : 0;
3463 743 : bool use_uncompressed_buffer = (!zused || zused > real_buffer_size);
3464 743 : CHECK_FORCE_UNCOMPRESSED_BUFFER();
3465 :
3466 : // setup payload header
3467 743 : cloudsync_payload_header header = {0};
3468 743 : uint32_t expanded_size = (use_uncompressed_buffer) ? 0 : real_buffer_size;
3469 743 : uint8_t version = payload->version ? payload->version : CLOUDSYNC_PAYLOAD_VERSION_LATEST;
3470 743 : cloudsync_payload_header_init(&header, version, expanded_size, payload->ncols, (uint32_t)payload->nrows, data->schema_hash);
3471 :
3472 : // if compression fails or if compressed size is bigger than original buffer, then use the uncompressed buffer
3473 743 : if (use_uncompressed_buffer) {
3474 44 : if (zbuffer) cloudsync_memory_free(zbuffer);
3475 44 : zbuffer = payload->buffer;
3476 44 : zused = real_buffer_size;
3477 44 : }
3478 :
3479 : // compute checksum of the buffer
3480 743 : uint64_t checksum = pk_checksum(zbuffer + header_size, zused);
3481 743 : cloudsync_payload_checksum_store(&header, checksum);
3482 :
3483 : // copy header and data to SQLite BLOB
3484 743 : memcpy(zbuffer, &header, sizeof(cloudsync_payload_header));
3485 743 : int blob_size = zused + sizeof(cloudsync_payload_header);
3486 743 : payload->bsize = blob_size;
3487 :
3488 : // cleanup memory
3489 743 : if (zbuffer != payload->buffer) {
3490 699 : cloudsync_memory_free (payload->buffer);
3491 699 : payload->buffer = zbuffer;
3492 699 : }
3493 :
3494 743 : return DBRES_OK;
3495 751 : }
3496 :
3497 751 : char *cloudsync_payload_blob (cloudsync_payload_context *payload, int64_t *blob_size, int64_t *nrows) {
3498 : DEBUG_FUNCTION("cloudsync_payload_blob");
3499 :
3500 751 : if (blob_size) *blob_size = (int64_t)payload->bsize;
3501 751 : if (nrows) *nrows = (int64_t)payload->nrows;
3502 751 : return payload->buffer;
3503 : }
3504 :
3505 462132 : static int cloudsync_payload_decode_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
3506 462132 : cloudsync_pk_decode_bind_context *decode_context = (cloudsync_pk_decode_bind_context*)xdata;
3507 462132 : int rc = pk_decode_bind_callback(decode_context->vm, index, type, ival, dval, pval);
3508 :
3509 462132 : if (rc == DBRES_OK) {
3510 : // the dbversion index is smaller than seq index, so it is processed first
3511 : // when processing the dbversion column: save the value to the tmp_dbversion field
3512 : // when processing the seq column: update the dbversion and seq fields only if the current dbversion is greater than the last max value
3513 462132 : switch (index) {
3514 : case CLOUDSYNC_PK_INDEX_TBL:
3515 51348 : if (type == DBTYPE_TEXT) {
3516 51348 : decode_context->tbl = pval;
3517 51348 : decode_context->tbl_len = ival;
3518 51348 : }
3519 51348 : break;
3520 : case CLOUDSYNC_PK_INDEX_PK:
3521 51348 : if (type == DBTYPE_BLOB) {
3522 51348 : decode_context->pk = pval;
3523 51348 : decode_context->pk_len = ival;
3524 51348 : }
3525 51348 : break;
3526 : case CLOUDSYNC_PK_INDEX_COLNAME:
3527 51348 : if (type == DBTYPE_TEXT) {
3528 51348 : decode_context->col_name = pval;
3529 51348 : decode_context->col_name_len = ival;
3530 51348 : }
3531 51348 : break;
3532 : case CLOUDSYNC_PK_INDEX_COLVERSION:
3533 51348 : if (type == DBTYPE_INTEGER) decode_context->col_version = ival;
3534 51348 : break;
3535 : case CLOUDSYNC_PK_INDEX_DBVERSION:
3536 51348 : if (type == DBTYPE_INTEGER) decode_context->db_version = ival;
3537 51348 : break;
3538 : case CLOUDSYNC_PK_INDEX_SITEID:
3539 51348 : if (type == DBTYPE_BLOB) {
3540 51348 : decode_context->site_id = pval;
3541 51348 : decode_context->site_id_len = ival;
3542 51348 : }
3543 51348 : break;
3544 : case CLOUDSYNC_PK_INDEX_CL:
3545 51348 : if (type == DBTYPE_INTEGER) decode_context->cl = ival;
3546 51348 : break;
3547 : case CLOUDSYNC_PK_INDEX_SEQ:
3548 51348 : if (type == DBTYPE_INTEGER) decode_context->seq = ival;
3549 51348 : break;
3550 : }
3551 462132 : }
3552 :
3553 462132 : return rc;
3554 : }
3555 :
3556 : typedef struct {
3557 : const char *tbl;
3558 : int64_t tbl_len;
3559 : const void *pk;
3560 : int64_t pk_len;
3561 : const char *col_name;
3562 : int64_t col_name_len;
3563 : const void *col_value;
3564 : int64_t col_value_len;
3565 : int64_t col_version;
3566 : int64_t db_version;
3567 : const void *site_id;
3568 : int64_t site_id_len;
3569 : int64_t cl;
3570 : int64_t seq;
3571 : } cloudsync_payload_fragment_row;
3572 :
3573 351 : static int cloudsync_payload_fragment_decode_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
3574 351 : UNUSED_PARAMETER(dval);
3575 351 : cloudsync_payload_fragment_row *row = (cloudsync_payload_fragment_row *)xdata;
3576 351 : switch (index) {
3577 : case CLOUDSYNC_PK_INDEX_TBL:
3578 39 : if (type != DBTYPE_TEXT) return DBRES_ERROR;
3579 39 : row->tbl = pval; row->tbl_len = ival;
3580 39 : break;
3581 : case CLOUDSYNC_PK_INDEX_PK:
3582 39 : if (type != DBTYPE_BLOB) return DBRES_ERROR;
3583 39 : row->pk = pval; row->pk_len = ival;
3584 39 : break;
3585 : case CLOUDSYNC_PK_INDEX_COLNAME:
3586 39 : if (type != DBTYPE_TEXT) return DBRES_ERROR;
3587 39 : row->col_name = pval; row->col_name_len = ival;
3588 39 : break;
3589 : case CLOUDSYNC_PK_INDEX_COLVALUE:
3590 39 : if (type != DBTYPE_BLOB) return DBRES_ERROR;
3591 39 : row->col_value = pval; row->col_value_len = ival;
3592 39 : break;
3593 : case CLOUDSYNC_PK_INDEX_COLVERSION:
3594 39 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3595 39 : row->col_version = ival;
3596 39 : break;
3597 : case CLOUDSYNC_PK_INDEX_DBVERSION:
3598 39 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3599 39 : row->db_version = ival;
3600 39 : break;
3601 : case CLOUDSYNC_PK_INDEX_SITEID:
3602 39 : if (type != DBTYPE_BLOB) return DBRES_ERROR;
3603 39 : row->site_id = pval; row->site_id_len = ival;
3604 39 : break;
3605 : case CLOUDSYNC_PK_INDEX_CL:
3606 39 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3607 39 : row->cl = ival;
3608 39 : break;
3609 : case CLOUDSYNC_PK_INDEX_SEQ:
3610 39 : if (type != DBTYPE_INTEGER) return DBRES_ERROR;
3611 39 : row->seq = ival;
3612 39 : break;
3613 : }
3614 351 : return DBRES_OK;
3615 351 : }
3616 :
3617 78 : static bool cloudsync_payload_is_hex (const char *value, size_t len) {
3618 1950 : for (size_t i = 0; i < len; ++i) {
3619 1872 : char c = value[i];
3620 1872 : if (!((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F'))) return false;
3621 1872 : }
3622 78 : return true;
3623 78 : }
3624 :
3625 117 : static bool cloudsync_payload_parse_u64_segment (const char *start, const char *end, uint64_t max_value, uint64_t *out, const char **next) {
3626 117 : if (!start || !end || start >= end) return false;
3627 117 : uint64_t value = 0;
3628 117 : const char *p = start;
3629 436 : while (p < end && *p >= '0' && *p <= '9') {
3630 319 : uint64_t digit = (uint64_t)(*p - '0');
3631 319 : if (value > (max_value - digit) / 10u) return false;
3632 319 : value = value * 10u + digit;
3633 319 : p++;
3634 : }
3635 117 : if (p == start || p >= end || *p != ':') return false;
3636 117 : *out = value;
3637 117 : *next = p + 1;
3638 117 : return true;
3639 117 : }
3640 :
3641 39 : static bool cloudsync_payload_fragment_parse_colname (const char *col_name, int64_t col_name_len,
3642 : char *value_id, size_t value_id_len,
3643 : char *checksum_hex, size_t checksum_hex_len,
3644 : int *part_index, int *part_count,
3645 : int64_t *total_size,
3646 : const char **base_col, int64_t *base_col_len) {
3647 39 : size_t prefix_len = strlen(CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX);
3648 39 : if (!col_name || col_name_len <= (int64_t)prefix_len) return false;
3649 39 : if (strncmp(col_name, CLOUDSYNC_PAYLOAD_FRAGMENT_PREFIX, prefix_len) != 0) return false;
3650 :
3651 39 : const char *p = col_name + prefix_len;
3652 39 : const char *end = col_name + col_name_len;
3653 39 : const char *sep = memchr(p, ':', (size_t)(end - p));
3654 39 : if (!sep || (size_t)(sep - p) + 1 > value_id_len) return false;
3655 39 : if ((sep - p) != 32 || !cloudsync_payload_is_hex(p, (size_t)(sep - p))) return false;
3656 39 : memcpy(value_id, p, (size_t)(sep - p));
3657 39 : value_id[sep - p] = 0;
3658 :
3659 39 : p = sep + 1;
3660 39 : sep = memchr(p, ':', (size_t)(end - p));
3661 39 : if (!sep || (size_t)(sep - p) + 1 > checksum_hex_len) return false;
3662 39 : if ((sep - p) != 16 || !cloudsync_payload_is_hex(p, (size_t)(sep - p))) return false;
3663 39 : memcpy(checksum_hex, p, (size_t)(sep - p));
3664 39 : checksum_hex[sep - p] = 0;
3665 :
3666 39 : const char *next = NULL;
3667 39 : uint64_t parsed = 0;
3668 39 : if (!cloudsync_payload_parse_u64_segment(sep + 1, end, INT_MAX, &parsed, &next)) return false;
3669 39 : *part_index = (int)parsed;
3670 :
3671 39 : if (!cloudsync_payload_parse_u64_segment(next, end, INT_MAX, &parsed, &next)) return false;
3672 39 : *part_count = (int)parsed;
3673 :
3674 39 : if (!cloudsync_payload_parse_u64_segment(next, end, INT64_MAX, &parsed, &next)) return false;
3675 39 : *total_size = (int64_t)parsed;
3676 :
3677 39 : *base_col = next;
3678 39 : *base_col_len = end - *base_col;
3679 39 : return (*part_count > 0 && *part_index < *part_count && *base_col_len > 0);
3680 39 : }
3681 :
3682 : typedef struct {
3683 : dbvm_t *vm;
3684 : int param_index;
3685 : } cloudsync_payload_bind_param_context;
3686 :
3687 11 : static int cloudsync_payload_bind_param_callback (void *xdata, int index, int type, int64_t ival, double dval, char *pval) {
3688 11 : UNUSED_PARAMETER(index);
3689 11 : cloudsync_payload_bind_param_context *ctx = (cloudsync_payload_bind_param_context *)xdata;
3690 11 : switch (type) {
3691 0 : case DBTYPE_INTEGER: return databasevm_bind_int(ctx->vm, ctx->param_index, ival);
3692 0 : case DBTYPE_FLOAT: return databasevm_bind_double(ctx->vm, ctx->param_index, dval);
3693 0 : case DBTYPE_NULL: return databasevm_bind_null(ctx->vm, ctx->param_index);
3694 2 : case DBTYPE_TEXT: return databasevm_bind_text(ctx->vm, ctx->param_index, pval, (int)ival);
3695 9 : case DBTYPE_BLOB: return databasevm_bind_blob(ctx->vm, ctx->param_index, pval, (uint64_t)ival);
3696 : }
3697 0 : return DBRES_MISUSE;
3698 11 : }
3699 :
3700 39 : static int cloudsync_payload_fragments_cleanup_stale (cloudsync_context *data) {
3701 : // Stale-fragment GC is pure maintenance (it removes incomplete fragment groups
3702 : // older than CLOUDSYNC_PAYLOAD_FRAGMENT_STALE_SECONDS), so it has no correctness
3703 : // deadline. It runs a full GROUP BY scan of the fragments table; calling it on
3704 : // every applied fragment would be O(n^2) for a heavily-fragmented value, since
3705 : // each fragment arrives as its own apply call. Throttle it to at most once per
3706 : // CLOUDSYNC_PAYLOAD_FRAGMENT_CLEANUP_MIN_INTERVAL per connection.
3707 39 : int64_t now = (int64_t)time(NULL);
3708 39 : if (data->last_fragment_cleanup != 0 &&
3709 33 : now - data->last_fragment_cleanup < CLOUDSYNC_PAYLOAD_FRAGMENT_CLEANUP_MIN_INTERVAL) {
3710 33 : return DBRES_OK;
3711 : }
3712 6 : data->last_fragment_cleanup = now;
3713 :
3714 6 : dbvm_t *vm = NULL;
3715 6 : int rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_CLEANUP_STALE, &vm, 0);
3716 6 : if (rc != DBRES_OK) return rc;
3717 6 : int64_t cutoff = now - CLOUDSYNC_PAYLOAD_FRAGMENT_STALE_SECONDS;
3718 6 : rc = databasevm_bind_int(vm, 1, cutoff);
3719 6 : if (rc == DBRES_OK) rc = databasevm_step(vm);
3720 6 : databasevm_finalize(vm);
3721 6 : return (rc == DBRES_DONE) ? DBRES_OK : rc;
3722 39 : }
3723 :
3724 11 : static int cloudsync_payload_apply_single_decoded_row (cloudsync_context *data,
3725 : const char *tbl, size_t tbl_len,
3726 : const char *pk, size_t pk_len,
3727 : const char *col_name, size_t col_name_len,
3728 : const char *encoded_value, size_t encoded_value_len,
3729 : int64_t col_version, int64_t db_version,
3730 : const char *site_id, size_t site_id_len,
3731 : int64_t cl, int64_t seq,
3732 : int *pnrows) {
3733 11 : int rc = DBRES_OK;
3734 11 : dbvm_t *vm = NULL;
3735 11 : bool in_savepoint = false;
3736 11 : merge_pending_batch batch = {0};
3737 :
3738 11 : rc = databasevm_prepare(data, SQL_CHANGES_INSERT_ROW, &vm, 0);
3739 11 : if (rc != DBRES_OK) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: error while compiling SQL statement", rc);
3740 :
3741 11 : rc = databasevm_bind_text(vm, 1, tbl, (int)tbl_len);
3742 11 : if (rc == DBRES_OK) rc = databasevm_bind_blob(vm, 2, pk, (uint64_t)pk_len);
3743 11 : if (rc == DBRES_OK) rc = databasevm_bind_text(vm, 3, col_name, (int)col_name_len);
3744 11 : if (rc == DBRES_OK) {
3745 11 : if (data->skip_decode_idx == CLOUDSYNC_PK_INDEX_COLVALUE) {
3746 0 : rc = databasevm_bind_blob(vm, 4, encoded_value, (uint64_t)encoded_value_len);
3747 0 : } else {
3748 11 : size_t seek = 0;
3749 11 : cloudsync_payload_bind_param_context bind_ctx = {.vm = vm, .param_index = 4};
3750 11 : int res = pk_decode((char *)encoded_value, encoded_value_len, 1, &seek, -1, cloudsync_payload_bind_param_callback, &bind_ctx);
3751 11 : if (res == -1 || seek != encoded_value_len) rc = cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 encoded value", DBRES_MISUSE);
3752 : }
3753 11 : }
3754 11 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 5, col_version);
3755 11 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 6, db_version);
3756 11 : if (rc == DBRES_OK) rc = databasevm_bind_blob(vm, 7, site_id, (uint64_t)site_id_len);
3757 11 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 8, cl);
3758 11 : if (rc == DBRES_OK) rc = databasevm_bind_int(vm, 9, seq);
3759 11 : if (rc != DBRES_OK) goto cleanup;
3760 :
3761 11 : if (!database_in_transaction(data)) {
3762 11 : rc = database_begin_savepoint(data, "cloudsync_payload_apply");
3763 11 : if (rc != DBRES_OK) goto cleanup;
3764 11 : in_savepoint = true;
3765 11 : }
3766 :
3767 11 : data->pending_batch = &batch;
3768 11 : rc = databasevm_step(vm);
3769 11 : if (rc == DBRES_DONE) rc = DBRES_OK;
3770 11 : if (rc != DBRES_OK) {
3771 0 : cloudsync_set_dberror(data);
3772 0 : goto cleanup;
3773 : }
3774 :
3775 11 : rc = merge_flush_pending(data);
3776 11 : if (rc != DBRES_OK) goto cleanup;
3777 11 : data->pending_batch = NULL;
3778 :
3779 11 : if (in_savepoint) {
3780 11 : rc = database_commit_savepoint(data, "cloudsync_payload_apply");
3781 11 : in_savepoint = false;
3782 11 : if (rc != DBRES_OK) goto cleanup;
3783 11 : }
3784 :
3785 : // Do NOT advance the receive cursor here: a v3 value carries a single
3786 : // (db_version, seq) that can be in the middle of its source db_version, and a
3787 : // db_version's chunks can span multiple /check artifacts. Advancing per value
3788 : // would leave the cursor mid-db_version. The durable cursor is advanced once,
3789 : // after the whole payload/stream is applied, via cloudsync_payload_apply's
3790 : // checkpoint argument. Record the last applied position for the
3791 : // CLOUDSYNC_CHECKPOINT_LAST_APPLIED mode.
3792 11 : if (db_version > data->apply_last_db_version ||
3793 0 : (db_version == data->apply_last_db_version && seq > data->apply_last_seq)) {
3794 11 : data->apply_last_db_version = db_version;
3795 11 : data->apply_last_seq = seq;
3796 11 : }
3797 :
3798 11 : if (pnrows) *pnrows += 1;
3799 :
3800 : cleanup:
3801 11 : if (rc != DBRES_OK && in_savepoint) database_rollback_savepoint(data, "cloudsync_payload_apply");
3802 11 : data->pending_batch = NULL;
3803 11 : merge_pending_free_entries(&batch);
3804 11 : if (batch.cached_vm) databasevm_finalize(batch.cached_vm);
3805 11 : if (batch.cached_col_names) cloudsync_memory_free(batch.cached_col_names);
3806 11 : if (batch.entries) cloudsync_memory_free(batch.entries);
3807 11 : if (vm) databasevm_finalize(vm);
3808 11 : return rc;
3809 11 : }
3810 :
3811 39 : static int cloudsync_payload_apply_reassembled_fragment (cloudsync_context *data, const char *value_id, const char *expected_checksum_hex, int *pnrows) {
3812 39 : int rc = DBRES_OK;
3813 39 : dbvm_t *vm = NULL;
3814 39 : char *value = NULL;
3815 39 : char *tbl = NULL, *col_name = NULL;
3816 39 : char *pk = NULL, *site_id = NULL;
3817 39 : size_t tbl_len = 0, col_name_len = 0, pk_len = 0, site_id_len = 0;
3818 39 : int64_t col_version = 0, db_version = 0, cl = 0, seq = 0;
3819 39 : int64_t total_size = 0, copied = 0;
3820 :
3821 39 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_COUNT, &vm, 0);
3822 39 : if (rc != DBRES_OK) return rc;
3823 39 : rc = databasevm_bind_text(vm, 1, value_id, -1);
3824 39 : if (rc != DBRES_OK) { databasevm_finalize(vm); return rc; }
3825 39 : rc = databasevm_step(vm);
3826 39 : if (rc != DBRES_ROW) { databasevm_finalize(vm); return DBRES_OK; }
3827 39 : int64_t have = database_column_int(vm, 0);
3828 39 : int64_t part_count_min = database_column_int(vm, 1);
3829 39 : int64_t part_count_max = database_column_int(vm, 2);
3830 39 : int64_t total_size_min = database_column_int(vm, 3);
3831 39 : int64_t total_size_max = database_column_int(vm, 4);
3832 39 : const char *checksum_min = database_column_text(vm, 5);
3833 39 : const char *checksum_max = database_column_text(vm, 6);
3834 39 : char checksum_min_copy[32] = {0};
3835 39 : char checksum_max_copy[32] = {0};
3836 39 : if (checksum_min) snprintf(checksum_min_copy, sizeof(checksum_min_copy), "%s", checksum_min);
3837 39 : if (checksum_max) snprintf(checksum_max_copy, sizeof(checksum_max_copy), "%s", checksum_max);
3838 39 : int64_t part_index_min = database_column_int(vm, 7);
3839 39 : int64_t part_index_max = database_column_int(vm, 8);
3840 39 : databasevm_finalize(vm);
3841 39 : vm = NULL;
3842 39 : if (have <= 0 || part_count_min <= 0 || have < part_count_max) return DBRES_OK;
3843 22 : if (part_count_min != part_count_max || total_size_min != total_size_max || !checksum_min_copy[0] || !checksum_max_copy[0] ||
3844 11 : strcmp(checksum_min_copy, checksum_max_copy) != 0 || strcmp(checksum_min_copy, expected_checksum_hex) != 0 ||
3845 11 : part_index_min != 0 || part_index_max != part_count_max - 1 || have != part_count_max) {
3846 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: inconsistent v3 fragments", DBRES_MISUSE);
3847 : }
3848 11 : total_size = total_size_max;
3849 :
3850 11 : value = cloudsync_memory_alloc((uint64_t)total_size);
3851 11 : if (!value) return DBRES_NOMEM;
3852 :
3853 11 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_SELECT, &vm, 0);
3854 11 : if (rc != DBRES_OK) goto cleanup;
3855 11 : rc = databasevm_bind_text(vm, 1, value_id, -1);
3856 11 : if (rc != DBRES_OK) goto cleanup;
3857 :
3858 11 : uint64_t checksum = 14695981039346656037ULL;
3859 48 : while ((rc = databasevm_step(vm)) == DBRES_ROW) {
3860 37 : size_t frag_len = 0;
3861 37 : const char *frag = database_column_blob(vm, 0, &frag_len);
3862 37 : if (copied + (int64_t)frag_len > total_size) { rc = DBRES_MISUSE; goto cleanup; }
3863 37 : memcpy(value + copied, frag, frag_len);
3864 37 : checksum = cloudsync_checksum_update(checksum, frag, frag_len);
3865 37 : copied += (int64_t)frag_len;
3866 :
3867 37 : if (!tbl) {
3868 11 : const char *t = database_column_text(vm, 1);
3869 11 : const char *c = database_column_text(vm, 3);
3870 11 : size_t pkl = 0, sidl = 0;
3871 11 : const char *p = database_column_blob(vm, 2, &pkl);
3872 11 : const char *sid = database_column_blob(vm, 6, &sidl);
3873 11 : tbl_len = (size_t)database_column_bytes(vm, 1);
3874 11 : col_name_len = (size_t)database_column_bytes(vm, 3);
3875 11 : pk_len = pkl;
3876 11 : site_id_len = sidl;
3877 11 : tbl = cloudsync_memory_alloc((uint64_t)tbl_len);
3878 11 : col_name = cloudsync_memory_alloc((uint64_t)col_name_len);
3879 11 : pk = cloudsync_memory_alloc((uint64_t)pk_len);
3880 11 : site_id = cloudsync_memory_alloc((uint64_t)site_id_len);
3881 11 : if (!tbl || !col_name || !pk || !site_id) { rc = DBRES_NOMEM; goto cleanup; }
3882 11 : memcpy(tbl, t, tbl_len);
3883 11 : memcpy(col_name, c, col_name_len);
3884 11 : memcpy(pk, p, pk_len);
3885 11 : memcpy(site_id, sid, site_id_len);
3886 11 : col_version = database_column_int(vm, 4);
3887 11 : db_version = database_column_int(vm, 5);
3888 11 : cl = database_column_int(vm, 7);
3889 11 : seq = database_column_int(vm, 8);
3890 11 : } else {
3891 26 : size_t pkl = 0, sidl = 0;
3892 26 : const char *t = database_column_text(vm, 1);
3893 26 : const char *c = database_column_text(vm, 3);
3894 26 : const char *p = database_column_blob(vm, 2, &pkl);
3895 26 : const char *sid = database_column_blob(vm, 6, &sidl);
3896 52 : if ((size_t)database_column_bytes(vm, 1) != tbl_len || memcmp(tbl, t, tbl_len) != 0 ||
3897 26 : pkl != pk_len || memcmp(pk, p, pk_len) != 0 ||
3898 26 : (size_t)database_column_bytes(vm, 3) != col_name_len || memcmp(col_name, c, col_name_len) != 0 ||
3899 26 : database_column_int(vm, 4) != col_version ||
3900 26 : database_column_int(vm, 5) != db_version ||
3901 26 : sidl != site_id_len || memcmp(site_id, sid, site_id_len) != 0 ||
3902 26 : database_column_int(vm, 7) != cl ||
3903 26 : database_column_int(vm, 8) != seq) {
3904 0 : rc = DBRES_MISUSE;
3905 0 : goto cleanup;
3906 : }
3907 : }
3908 : }
3909 11 : if (rc == DBRES_DONE) rc = DBRES_OK;
3910 11 : if (rc != DBRES_OK) goto cleanup;
3911 11 : if (copied != total_size) { rc = DBRES_MISUSE; goto cleanup; }
3912 : char checksum_hex[17];
3913 11 : snprintf(checksum_hex, sizeof(checksum_hex), "%016" PRIx64, checksum);
3914 11 : if (strcmp(checksum_hex, expected_checksum_hex) != 0) { rc = DBRES_MISUSE; goto cleanup; }
3915 11 : databasevm_finalize(vm);
3916 11 : vm = NULL;
3917 :
3918 22 : rc = cloudsync_payload_apply_single_decoded_row(data, tbl, tbl_len, pk, pk_len, col_name, col_name_len,
3919 11 : value, (size_t)total_size, col_version, db_version,
3920 11 : site_id, site_id_len, cl, seq, pnrows);
3921 11 : if (rc != DBRES_OK) goto cleanup;
3922 :
3923 11 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_DELETE, &vm, 0);
3924 22 : if (rc == DBRES_OK) {
3925 11 : databasevm_bind_text(vm, 1, value_id, -1);
3926 11 : int step_rc = databasevm_step(vm);
3927 11 : if (step_rc == DBRES_DONE) rc = DBRES_OK;
3928 11 : }
3929 :
3930 : cleanup:
3931 11 : if (vm) databasevm_finalize(vm);
3932 11 : if (value) cloudsync_memory_free(value);
3933 11 : if (tbl) cloudsync_memory_free(tbl);
3934 11 : if (col_name) cloudsync_memory_free(col_name);
3935 11 : if (pk) cloudsync_memory_free(pk);
3936 11 : if (site_id) cloudsync_memory_free(site_id);
3937 11 : return rc;
3938 39 : }
3939 :
3940 39 : static int cloudsync_payload_apply_fragment_row (cloudsync_context *data, cloudsync_payload_fragment_row *row, int *pnrows) {
3941 : char value_id[64];
3942 : char checksum_hex[17];
3943 39 : int part_index = 0, part_count = 0;
3944 39 : int64_t total_size = 0;
3945 39 : const char *base_col = NULL;
3946 39 : int64_t base_col_len = 0;
3947 78 : if (!row || !row->tbl || row->tbl_len <= 0 || !row->pk || row->pk_len <= 0 ||
3948 39 : !row->col_name || row->col_name_len <= 0 || !row->col_value || row->col_value_len <= 0 ||
3949 39 : !row->site_id || row->site_id_len <= 0) {
3950 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 payload row", DBRES_MISUSE);
3951 : }
3952 78 : if (!cloudsync_payload_fragment_parse_colname(row->col_name, row->col_name_len, value_id, sizeof(value_id),
3953 39 : checksum_hex, sizeof(checksum_hex),
3954 : &part_index, &part_count, &total_size, &base_col, &base_col_len)) {
3955 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 fragment metadata", DBRES_MISUSE);
3956 : }
3957 :
3958 39 : uint64_t value_checksum = strtoull(checksum_hex, NULL, 16);
3959 : char expected_value_id[33];
3960 78 : cloudsync_payload_fragment_value_id(expected_value_id, row->tbl, (int)row->tbl_len, row->pk, (int)row->pk_len,
3961 39 : base_col, (int)base_col_len, row->col_version, row->db_version,
3962 39 : row->site_id, (int)row->site_id_len, row->cl, row->seq,
3963 39 : value_checksum, total_size);
3964 39 : if (strcmp(value_id, expected_value_id) != 0) {
3965 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 fragment identity", DBRES_MISUSE);
3966 : }
3967 :
3968 : // the fragments table is guaranteed by dbutils_settings_init; no DDL here
3969 : // because the apply path runs under sync-only credentials on server nodes
3970 39 : int rc = cloudsync_payload_fragments_cleanup_stale(data);
3971 39 : if (rc != DBRES_OK) return rc;
3972 :
3973 39 : dbvm_t *vm = NULL;
3974 39 : rc = databasevm_prepare(data, SQL_PAYLOAD_FRAGMENTS_UPSERT, &vm, 0);
3975 39 : if (rc != DBRES_OK) return rc;
3976 39 : databasevm_bind_text(vm, 1, value_id, -1);
3977 39 : databasevm_bind_int(vm, 2, part_index);
3978 39 : databasevm_bind_int(vm, 3, part_count);
3979 39 : databasevm_bind_int(vm, 4, total_size);
3980 39 : databasevm_bind_text(vm, 5, checksum_hex, -1);
3981 39 : databasevm_bind_int(vm, 6, (int64_t)time(NULL));
3982 39 : databasevm_bind_text(vm, 7, row->tbl, (int)row->tbl_len);
3983 39 : databasevm_bind_blob(vm, 8, row->pk, (uint64_t)row->pk_len);
3984 39 : databasevm_bind_text(vm, 9, base_col, (int)base_col_len);
3985 39 : databasevm_bind_int(vm, 10, row->col_version);
3986 39 : databasevm_bind_int(vm, 11, row->db_version);
3987 39 : databasevm_bind_blob(vm, 12, row->site_id, (uint64_t)row->site_id_len);
3988 39 : databasevm_bind_int(vm, 13, row->cl);
3989 39 : databasevm_bind_int(vm, 14, row->seq);
3990 39 : databasevm_bind_blob(vm, 15, row->col_value, (uint64_t)row->col_value_len);
3991 39 : rc = databasevm_step(vm);
3992 39 : databasevm_finalize(vm);
3993 39 : if (rc == DBRES_DONE) rc = DBRES_OK;
3994 39 : if (rc != DBRES_OK) return rc;
3995 :
3996 39 : return cloudsync_payload_apply_reassembled_fragment(data, value_id, checksum_hex, pnrows);
3997 39 : }
3998 :
3999 : // #ifndef CLOUDSYNC_OMIT_RLS_VALIDATION
4000 :
4001 : // Advance the durable receive cursor (check_dbversion/check_seq) after a payload
4002 : // (or a fully-applied chunk stream) has been applied. See the checkpoint-mode
4003 : // documentation on cloudsync_payload_apply in cloudsync.h. The advance is
4004 : // strictly monotonic so re-delivered rows never regress the cursor.
4005 746 : static void cloudsync_payload_apply_checkpoint (cloudsync_context *data, int64_t checkpoint_db_version, int64_t checkpoint_seq) {
4006 : int64_t target_db_version;
4007 : int64_t target_seq;
4008 :
4009 746 : if (checkpoint_db_version == CLOUDSYNC_CHECKPOINT_NONE) return;
4010 743 : if (checkpoint_db_version == CLOUDSYNC_CHECKPOINT_LAST_APPLIED) {
4011 : // Nothing applied -> nothing to checkpoint.
4012 742 : if (data->apply_last_db_version < 0) return;
4013 714 : target_db_version = data->apply_last_db_version;
4014 714 : target_seq = data->apply_last_seq;
4015 714 : } else {
4016 1 : target_db_version = checkpoint_db_version;
4017 1 : target_seq = checkpoint_seq;
4018 : }
4019 :
4020 715 : int64_t cur_db_version = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION);
4021 715 : int64_t cur_seq = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_CHECK_SEQ);
4022 :
4023 : // monotonic guard: never move the cursor backwards
4024 715 : if (target_db_version < cur_db_version) return;
4025 572 : if (target_db_version == cur_db_version && target_seq <= cur_seq) return;
4026 :
4027 : char buf[256];
4028 489 : snprintf(buf, sizeof(buf), "%" PRId64, target_db_version);
4029 489 : dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION, buf);
4030 489 : if (target_seq != cur_seq) {
4031 323 : snprintf(buf, sizeof(buf), "%" PRId64, target_seq);
4032 323 : dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_SEQ, buf);
4033 323 : }
4034 746 : }
4035 :
4036 751 : int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *pnrows, int64_t checkpoint_db_version, int64_t checkpoint_seq) {
4037 : // Guard against calling payload_apply before cloudsync_init: without this,
4038 : // the settings lookups at the top of this function would each emit a
4039 : // "no such table: cloudsync_settings" debug line, control would fall
4040 : // through to the meta-table insert, and the function would ultimately
4041 : // return an error with an empty errmsg — SQLite then surfaces that as
4042 : // the confusing "Runtime error: not an error".
4043 751 : if (!cloudsync_context_is_initialized(data)) {
4044 1 : return cloudsync_set_error(data,
4045 : "cloudsync is not initialized: call SELECT cloudsync_init('<table_name>') "
4046 : "to enable sync on a table before calling cloudsync_payload_apply().",
4047 : DBRES_MISUSE);
4048 : }
4049 :
4050 : // sanity check
4051 750 : if (blen < (int)sizeof(cloudsync_payload_header)) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid payload length", DBRES_MISUSE);
4052 :
4053 : // track the last (db_version, seq) applied by this call so the receive
4054 : // checkpoint can be computed once, after the whole payload is applied
4055 750 : data->apply_last_db_version = -1;
4056 750 : data->apply_last_seq = -1;
4057 :
4058 : // decode header
4059 : cloudsync_payload_header header;
4060 750 : memcpy(&header, payload, sizeof(cloudsync_payload_header));
4061 :
4062 750 : header.signature = ntohl(header.signature);
4063 750 : header.expanded_size = ntohl(header.expanded_size);
4064 750 : header.ncols = ntohs(header.ncols);
4065 750 : header.nrows = ntohl(header.nrows);
4066 750 : header.schema_hash = ntohll(header.schema_hash);
4067 :
4068 : // compare schema_hash only if not disabled and if the received payload was created with the current header version
4069 : // to avoid schema hash mismatch when processed by a peer with a different extension version during software updates.
4070 750 : if (dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_SKIP_SCHEMA_HASH_CHECK) == 0 && header.version == CLOUDSYNC_PAYLOAD_VERSION_LATEST ) {
4071 711 : if (header.schema_hash != data->schema_hash) {
4072 4 : if (!database_check_schema_hash(data, header.schema_hash)) {
4073 : char buffer[1024];
4074 2 : snprintf(buffer, sizeof(buffer), "Cannot apply the received payload because the schema hash is unknown %" PRIu64 ".", header.schema_hash);
4075 2 : return cloudsync_set_error(data, buffer, DBRES_MISUSE);
4076 : }
4077 2 : }
4078 709 : }
4079 :
4080 : // sanity check header
4081 748 : if ((header.signature != CLOUDSYNC_PAYLOAD_SIGNATURE) || (header.ncols == 0)) {
4082 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid signature or column size", DBRES_MISUSE);
4083 : }
4084 748 : if (header.version < CLOUDSYNC_PAYLOAD_VERSION_1 || header.version > CLOUDSYNC_PAYLOAD_VERSION_3) {
4085 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unsupported payload version", DBRES_MISUSE);
4086 : }
4087 :
4088 748 : const char *buffer = payload + sizeof(cloudsync_payload_header);
4089 748 : size_t buf_len = (size_t)blen - sizeof(cloudsync_payload_header);
4090 :
4091 : // sanity check checksum (only if version is >= 2)
4092 748 : if (header.version >= CLOUDSYNC_PAYLOAD_MIN_VERSION_WITH_CHECKSUM) {
4093 748 : uint64_t checksum = pk_checksum(buffer, buf_len);
4094 748 : if (cloudsync_payload_checksum_verify(&header, checksum) == false) {
4095 1 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid checksum", DBRES_MISUSE);
4096 : }
4097 747 : }
4098 :
4099 : // check if payload is compressed
4100 747 : char *clone = NULL;
4101 747 : if (header.expanded_size != 0) {
4102 692 : clone = (char *)cloudsync_memory_alloc(header.expanded_size);
4103 692 : if (!clone) return cloudsync_set_error(data, "Unable to allocate memory to uncompress payload", DBRES_NOMEM);
4104 :
4105 692 : int lz4_rc = LZ4_decompress_safe(buffer, clone, (int)buf_len, (int)header.expanded_size);
4106 692 : if (lz4_rc <= 0 || (uint32_t)lz4_rc != header.expanded_size) {
4107 0 : if (clone) cloudsync_memory_free(clone);
4108 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to decompress BLOB", DBRES_MISUSE);
4109 : }
4110 :
4111 692 : buffer = (const char *)clone;
4112 692 : buf_len = (size_t)header.expanded_size;
4113 692 : }
4114 :
4115 747 : if (header.version == CLOUDSYNC_PAYLOAD_VERSION_3) {
4116 39 : int rc = DBRES_OK;
4117 39 : int applied_rows = 0;
4118 39 : if (header.ncols != CLOUDSYNC_CHANGES_NCOLS) {
4119 0 : if (clone) cloudsync_memory_free(clone);
4120 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 column count", DBRES_MISUSE);
4121 : }
4122 78 : for (uint32_t i = 0; i < header.nrows; ++i) {
4123 39 : size_t seek = 0;
4124 39 : cloudsync_payload_fragment_row row = {0};
4125 39 : int res = pk_decode((char *)buffer, buf_len, header.ncols, &seek, -1,
4126 : cloudsync_payload_fragment_decode_callback, &row);
4127 39 : if (res == -1 || seek == 0 || seek > buf_len) {
4128 0 : rc = cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid v3 payload row", DBRES_MISUSE);
4129 0 : break;
4130 : }
4131 39 : int n = 0;
4132 39 : rc = cloudsync_payload_apply_fragment_row(data, &row, &n);
4133 39 : if (rc != DBRES_OK) break;
4134 39 : applied_rows += n;
4135 39 : buffer += seek;
4136 39 : buf_len -= seek;
4137 39 : }
4138 39 : if (clone) cloudsync_memory_free(clone);
4139 39 : if (pnrows) *pnrows = applied_rows;
4140 : // Advance the receive cursor only after the whole payload is applied,
4141 : // gated on the caller-supplied checkpoint (a non-final chunk passes
4142 : // CLOUDSYNC_CHECKPOINT_NONE and leaves the cursor untouched).
4143 39 : if (rc == DBRES_OK) cloudsync_payload_apply_checkpoint(data, checkpoint_db_version, checkpoint_seq);
4144 39 : return rc;
4145 : }
4146 :
4147 : // precompile the insert statement
4148 708 : dbvm_t *vm = NULL;
4149 708 : int rc = databasevm_prepare(data, SQL_CHANGES_INSERT_ROW, &vm, 0);
4150 708 : if (rc != DBRES_OK) {
4151 0 : if (clone) cloudsync_memory_free(clone);
4152 0 : return cloudsync_set_error(data, "Error on cloudsync_payload_apply: error while compiling SQL statement", rc);
4153 : }
4154 :
4155 : // process buffer, one row at a time
4156 708 : uint16_t ncols = header.ncols;
4157 708 : uint32_t nrows = header.nrows;
4158 708 : int64_t last_payload_db_version = -1;
4159 708 : cloudsync_pk_decode_bind_context decoded_context = {.vm = vm};
4160 :
4161 : // Initialize deferred column-batch merge
4162 708 : merge_pending_batch batch = {0};
4163 708 : data->pending_batch = &batch;
4164 708 : bool in_savepoint = false;
4165 708 : const void *last_pk = NULL;
4166 708 : int64_t last_pk_len = 0;
4167 708 : const char *last_tbl = NULL;
4168 708 : int64_t last_tbl_len = 0;
4169 :
4170 52056 : for (uint32_t i=0; i<nrows; ++i) {
4171 51348 : size_t seek = 0;
4172 51348 : int res = pk_decode((char *)buffer, buf_len, ncols, &seek, data->skip_decode_idx, cloudsync_payload_decode_callback, &decoded_context);
4173 51348 : if (res == -1) {
4174 0 : merge_flush_pending(data);
4175 0 : data->pending_batch = NULL;
4176 0 : if (batch.cached_vm) { databasevm_finalize(batch.cached_vm); batch.cached_vm = NULL; }
4177 0 : if (batch.cached_col_names) { cloudsync_memory_free(batch.cached_col_names); batch.cached_col_names = NULL; }
4178 0 : if (batch.entries) { cloudsync_memory_free(batch.entries); batch.entries = NULL; }
4179 0 : if (in_savepoint) database_rollback_savepoint(data, "cloudsync_payload_apply");
4180 0 : rc = DBRES_ERROR;
4181 0 : goto cleanup;
4182 : }
4183 :
4184 : // Detect PK/table/db_version boundary to flush pending batch
4185 101988 : bool pk_changed = (last_pk != NULL &&
4186 50640 : (last_pk_len != decoded_context.pk_len ||
4187 48720 : memcmp(last_pk, decoded_context.pk, last_pk_len) != 0));
4188 101988 : bool tbl_changed = (last_tbl != NULL &&
4189 50640 : (last_tbl_len != decoded_context.tbl_len ||
4190 50568 : memcmp(last_tbl, decoded_context.tbl, last_tbl_len) != 0));
4191 51348 : bool db_version_changed = (last_payload_db_version != decoded_context.db_version);
4192 :
4193 : // Flush pending batch before any boundary change
4194 51348 : if (pk_changed || tbl_changed || db_version_changed) {
4195 19954 : int flush_rc = merge_flush_pending(data);
4196 19954 : if (flush_rc != DBRES_OK) {
4197 1 : rc = flush_rc;
4198 : // continue processing remaining rows
4199 1 : }
4200 19954 : }
4201 :
4202 : // Per-db_version savepoints group rows with the same source db_version
4203 : // into one transaction. In SQLite autocommit mode, the RELEASE triggers
4204 : // the commit hook which bumps data->db_version and resets seq, ensuring
4205 : // unique (db_version, seq) tuples across groups. In PostgreSQL SPI,
4206 : // database_in_transaction() is always true so this block is inactive —
4207 : // the inner per-PK savepoint in merge_flush_pending handles RLS instead.
4208 51348 : if (in_savepoint && db_version_changed) {
4209 4299 : rc = database_commit_savepoint(data, "cloudsync_payload_apply");
4210 4299 : if (rc != DBRES_OK) {
4211 0 : merge_pending_free_entries(&batch);
4212 0 : data->pending_batch = NULL;
4213 0 : cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to release a savepoint", rc);
4214 0 : goto cleanup;
4215 : }
4216 4299 : in_savepoint = false;
4217 4299 : }
4218 :
4219 51348 : if (!in_savepoint && db_version_changed && !database_in_transaction(data)) {
4220 5007 : rc = database_begin_savepoint(data, "cloudsync_payload_apply");
4221 5007 : if (rc != DBRES_OK) {
4222 0 : merge_pending_free_entries(&batch);
4223 0 : data->pending_batch = NULL;
4224 0 : cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to start a transaction", rc);
4225 0 : goto cleanup;
4226 : }
4227 5007 : in_savepoint = true;
4228 5007 : }
4229 :
4230 : // Track db_version for batch-flush boundary detection
4231 51348 : if (db_version_changed) {
4232 5007 : last_payload_db_version = decoded_context.db_version;
4233 5007 : }
4234 :
4235 : // Update PK/table tracking
4236 51348 : last_pk = decoded_context.pk;
4237 51348 : last_pk_len = decoded_context.pk_len;
4238 51348 : last_tbl = decoded_context.tbl;
4239 51348 : last_tbl_len = decoded_context.tbl_len;
4240 :
4241 51348 : rc = databasevm_step(vm);
4242 51348 : if (rc != DBRES_DONE) {
4243 : // don't "break;", the error can be due to a RLS policy.
4244 : // in case of error we try to apply the following changes
4245 2 : }
4246 :
4247 51348 : buffer += seek;
4248 51348 : buf_len -= seek;
4249 51348 : dbvm_reset(vm);
4250 51348 : }
4251 :
4252 : // Final flush after loop
4253 : {
4254 708 : int flush_rc = merge_flush_pending(data);
4255 708 : if (flush_rc != DBRES_OK && rc == DBRES_OK) rc = flush_rc;
4256 : }
4257 708 : data->pending_batch = NULL;
4258 :
4259 708 : if (in_savepoint) {
4260 708 : int rc1 = database_commit_savepoint(data, "cloudsync_payload_apply");
4261 708 : if (rc1 != DBRES_OK) rc = rc1;
4262 708 : }
4263 :
4264 : // save last error (unused if function returns OK)
4265 708 : if (rc != DBRES_OK && rc != DBRES_DONE) {
4266 1 : cloudsync_set_dberror(data);
4267 1 : }
4268 :
4269 708 : if (rc == DBRES_DONE) rc = DBRES_OK;
4270 1415 : if (rc == DBRES_OK) {
4271 : // Record the last applied (db_version, seq) and advance the receive cursor
4272 : // once, gated on the caller-supplied checkpoint. A non-final chunk passes
4273 : // CLOUDSYNC_CHECKPOINT_NONE so the cursor never lands mid-db_version.
4274 707 : if (decoded_context.db_version > data->apply_last_db_version ||
4275 0 : (decoded_context.db_version == data->apply_last_db_version && decoded_context.seq > data->apply_last_seq)) {
4276 707 : data->apply_last_db_version = decoded_context.db_version;
4277 707 : data->apply_last_seq = decoded_context.seq;
4278 707 : }
4279 707 : cloudsync_payload_apply_checkpoint(data, checkpoint_db_version, checkpoint_seq);
4280 707 : }
4281 :
4282 : cleanup:
4283 : // cleanup merge_pending_batch
4284 708 : if (batch.cached_vm) { databasevm_finalize(batch.cached_vm); batch.cached_vm = NULL; }
4285 708 : if (batch.cached_col_names) { cloudsync_memory_free(batch.cached_col_names); batch.cached_col_names = NULL; }
4286 708 : if (batch.entries) { cloudsync_memory_free(batch.entries); batch.entries = NULL; }
4287 :
4288 : // cleanup vm
4289 708 : if (vm) databasevm_finalize(vm);
4290 :
4291 : // cleanup memory
4292 708 : if (clone) cloudsync_memory_free(clone);
4293 :
4294 : // error already saved in (save last error)
4295 708 : if (rc != DBRES_OK) return rc;
4296 :
4297 : // return the number of processed rows
4298 707 : if (pnrows) *pnrows = nrows;
4299 707 : return DBRES_OK;
4300 751 : }
4301 :
4302 : // MARK: - Payload load/store -
4303 :
4304 0 : int cloudsync_payload_get (cloudsync_context *data, char **blob, int *blob_size, int *db_version, int64_t *new_db_version) {
4305 : // retrieve current db_version and seq
4306 0 : *db_version = dbutils_settings_get_int_value(data, CLOUDSYNC_KEY_SEND_DBVERSION);
4307 0 : if (*db_version < 0) return DBRES_ERROR;
4308 :
4309 : // retrieve BLOB
4310 : char sql[1024];
4311 0 : snprintf(sql, sizeof(sql), "WITH max_db_version AS (SELECT MAX(db_version) AS max_db_version FROM cloudsync_changes WHERE site_id=cloudsync_siteid()) "
4312 : "SELECT * FROM (SELECT cloudsync_payload_encode(tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq) AS payload, max_db_version AS max_db_version FROM cloudsync_changes, max_db_version WHERE site_id=cloudsync_siteid() AND db_version>%d) WHERE payload IS NOT NULL", *db_version);
4313 :
4314 0 : int64_t len = 0;
4315 0 : int rc = database_select_blob_int(data, sql, blob, &len, new_db_version);
4316 0 : *blob_size = (int)len;
4317 0 : if (rc != DBRES_OK) return rc;
4318 :
4319 : // exit if there is no data to send
4320 0 : if (*blob == NULL || *blob_size == 0) return DBRES_OK;
4321 0 : return rc;
4322 0 : }
4323 :
4324 : #ifdef CLOUDSYNC_DESKTOP_OS
4325 0 : int cloudsync_payload_save (cloudsync_context *data, const char *payload_path, int *size) {
4326 : DEBUG_FUNCTION("cloudsync_payload_save");
4327 :
4328 : // silently delete any other payload with the same name
4329 0 : cloudsync_file_delete(payload_path);
4330 :
4331 : // retrieve payload
4332 0 : char *blob = NULL;
4333 0 : int blob_size = 0, db_version = 0;
4334 0 : int64_t new_db_version = 0;
4335 0 : int rc = cloudsync_payload_get(data, &blob, &blob_size, &db_version, &new_db_version);
4336 0 : if (rc != DBRES_OK) {
4337 0 : if (db_version < 0) return cloudsync_set_error(data, "Unable to retrieve db_version", rc);
4338 0 : return cloudsync_set_error(data, "Unable to retrieve changes in cloudsync_payload_save", rc);
4339 : }
4340 :
4341 : // exit if there is no data to save
4342 0 : if (blob == NULL || blob_size == 0) {
4343 0 : if (size) *size = 0;
4344 0 : return DBRES_OK;
4345 : }
4346 :
4347 : // write payload to file
4348 0 : bool res = cloudsync_file_write(payload_path, blob, (size_t)blob_size);
4349 0 : cloudsync_memory_free(blob);
4350 0 : if (res == false) {
4351 0 : return cloudsync_set_error(data, "Unable to write payload to file path", DBRES_IOERR);
4352 : }
4353 :
4354 : // returns blob size
4355 0 : if (size) *size = blob_size;
4356 0 : return DBRES_OK;
4357 0 : }
4358 : #endif
4359 :
4360 : // MARK: - Core -
4361 :
4362 312 : int cloudsync_table_sanity_check (cloudsync_context *data, const char *name, CLOUDSYNC_INIT_FLAG init_flags) {
4363 : DEBUG_DBFUNCTION("cloudsync_table_sanity_check %s", name);
4364 : char buffer[2048];
4365 :
4366 : // sanity check table name
4367 312 : if (name == NULL) {
4368 1 : return cloudsync_set_error(data, "cloudsync_init requires a non-null table parameter", DBRES_ERROR);
4369 : }
4370 :
4371 : // avoid allocating heap memory for SQL statements by setting a maximum length of 512 characters
4372 : // for table names. This limit is reasonable and helps prevent memory management issues.
4373 311 : const size_t maxlen = CLOUDSYNC_MAX_TABLENAME_LEN;
4374 311 : if (strlen(name) > maxlen) {
4375 1 : snprintf(buffer, sizeof(buffer), "Table name cannot be longer than %d characters", (int)maxlen);
4376 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4377 : }
4378 :
4379 : // check if already initialized
4380 310 : cloudsync_table_context *table = table_lookup(data, name);
4381 310 : if (table) return DBRES_OK;
4382 :
4383 : // check if table exists
4384 306 : if (database_table_exists(data, name, cloudsync_schema(data)) == false) {
4385 2 : snprintf(buffer, sizeof(buffer), "Table %s does not exist", name);
4386 2 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4387 : }
4388 :
4389 : // no more than 128 columns can be used as a composite primary key (SQLite hard limit)
4390 304 : int npri_keys = database_count_pk(data, name, false, cloudsync_schema(data));
4391 304 : if (npri_keys < 0) return cloudsync_set_dberror(data);
4392 304 : if (npri_keys > 128) return cloudsync_set_error(data, "No more than 128 columns can be used to form a composite primary key", DBRES_ERROR);
4393 :
4394 : #if CLOUDSYNC_DISABLE_ROWIDONLY_TABLES
4395 : // if count == 0 means that rowid will be used as primary key (BTW: very bad choice for the user)
4396 304 : if (npri_keys == 0) {
4397 1 : snprintf(buffer, sizeof(buffer), "Rowid only tables are not supported, all primary keys must be explicitly set and declared as NOT NULL (table %s)", name);
4398 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4399 : }
4400 : #endif
4401 :
4402 303 : bool skip_int_pk_check = (init_flags & CLOUDSYNC_INIT_FLAG_SKIP_INT_PK_CHECK) != 0;
4403 303 : if (!skip_int_pk_check) {
4404 240 : if (npri_keys == 1) {
4405 : // the affinity of a column is determined by the declared type of the column,
4406 : // according to the following rules in the order shown:
4407 : // 1. If the declared type contains the string "INT" then it is assigned INTEGER affinity.
4408 152 : int npri_keys_int = database_count_int_pk(data, name, cloudsync_schema(data));
4409 152 : if (npri_keys_int < 0) return cloudsync_set_dberror(data);
4410 152 : if (npri_keys == npri_keys_int) {
4411 1 : snprintf(buffer, sizeof(buffer), "Table %s uses a single-column INTEGER primary key. For CRDT replication, primary keys must be globally unique. Consider using a TEXT primary key with UUIDs or ULID to avoid conflicts across nodes. If you understand the risk and still want to use this INTEGER primary key, set the third argument of the cloudsync_init function to 1 to skip this check.", name);
4412 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4413 : }
4414 :
4415 151 : }
4416 239 : }
4417 :
4418 : // if user declared explicit primary key(s) then make sure they are all declared as NOT NULL
4419 : #if CLOUDSYNC_CHECK_NOTNULL_PRIKEYS
4420 : bool skip_notnull_prikeys_check = (init_flags & CLOUDSYNC_INIT_FLAG_SKIP_NOT_NULL_PRIKEYS_CHECK) != 0;
4421 : if (!skip_notnull_prikeys_check) {
4422 : if (npri_keys > 0) {
4423 : int npri_keys_notnull = database_count_pk(data, name, true, cloudsync_schema(data));
4424 : if (npri_keys_notnull < 0) return cloudsync_set_dberror(data);
4425 : if (npri_keys != npri_keys_notnull) {
4426 : snprintf(buffer, sizeof(buffer), "All primary keys must be explicitly declared as NOT NULL (table %s)", name);
4427 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4428 : }
4429 : }
4430 : }
4431 : #endif
4432 :
4433 : // check for columns declared as NOT NULL without a DEFAULT value.
4434 : // Otherwise, col_merge_stmt would fail if changes to other columns are inserted first.
4435 302 : bool skip_notnull_default_check = (init_flags & CLOUDSYNC_INIT_FLAG_SKIP_NOT_NULL_DEFAULT_CHECK) != 0;
4436 302 : if (!skip_notnull_default_check) {
4437 302 : int n_notnull_nodefault = database_count_notnull_without_default(data, name, cloudsync_schema(data));
4438 302 : if (n_notnull_nodefault < 0) return cloudsync_set_dberror(data);
4439 302 : if (n_notnull_nodefault > 0) {
4440 0 : snprintf(buffer, sizeof(buffer), "All non-primary key columns declared as NOT NULL must have a DEFAULT value. (table %s)", name);
4441 0 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4442 : }
4443 302 : }
4444 :
4445 302 : return DBRES_OK;
4446 312 : }
4447 :
4448 4 : int cloudsync_cleanup_internal (cloudsync_context *data, cloudsync_table_context *table) {
4449 4 : if (cloudsync_context_init(data) == NULL) return DBRES_MISUSE;
4450 :
4451 : // drop meta-table
4452 4 : const char *table_name = table->name;
4453 4 : char *sql = cloudsync_memory_mprintf(SQL_DROP_CLOUDSYNC_TABLE, table->meta_ref);
4454 4 : int rc = database_exec(data, sql);
4455 4 : cloudsync_memory_free(sql);
4456 4 : if (rc != DBRES_OK) {
4457 : char buffer[1024];
4458 0 : snprintf(buffer, sizeof(buffer), "Unable to drop cloudsync table %s_cloudsync in cloudsync_cleanup", table_name);
4459 0 : return cloudsync_set_error(data, buffer, rc);
4460 : }
4461 :
4462 : // drop blocks table if this table has block LWW columns
4463 4 : if (table->blocks_ref) {
4464 1 : sql = cloudsync_memory_mprintf(SQL_DROP_CLOUDSYNC_TABLE, table->blocks_ref);
4465 1 : rc = database_exec(data, sql);
4466 1 : cloudsync_memory_free(sql);
4467 1 : if (rc != DBRES_OK) {
4468 : char buffer[1024];
4469 0 : snprintf(buffer, sizeof(buffer), "Unable to drop blocks table %s_cloudsync_blocks in cloudsync_cleanup", table_name);
4470 0 : return cloudsync_set_error(data, buffer, rc);
4471 : }
4472 1 : }
4473 :
4474 : // drop original triggers
4475 4 : rc = database_delete_triggers(data, table_name);
4476 4 : if (rc != DBRES_OK) {
4477 : char buffer[1024];
4478 0 : snprintf(buffer, sizeof(buffer), "Unable to delete triggers for table %s", table_name);
4479 0 : return cloudsync_set_error(data, buffer, rc);
4480 : }
4481 :
4482 : // remove all table related settings
4483 4 : dbutils_table_settings_set_key_value(data, table_name, NULL, NULL, NULL);
4484 4 : return DBRES_OK;
4485 4 : }
4486 :
4487 4 : int cloudsync_cleanup (cloudsync_context *data, const char *table_name) {
4488 4 : cloudsync_table_context *table = table_lookup(data, table_name);
4489 4 : if (!table) return DBRES_OK;
4490 :
4491 : // TODO: check what happen if cloudsync_cleanup_internal failes (not eveything dropped) and the table is still in memory?
4492 :
4493 4 : int rc = cloudsync_cleanup_internal(data, table);
4494 4 : if (rc != DBRES_OK) return rc;
4495 :
4496 4 : int counter = table_remove(data, table);
4497 4 : table_free(table);
4498 :
4499 4 : if (counter == 0) {
4500 : // cleanup database on last table
4501 2 : cloudsync_reset_siteid(data);
4502 2 : dbutils_settings_cleanup(data);
4503 2 : } else {
4504 2 : if (database_internal_table_exists(data, CLOUDSYNC_TABLE_SETTINGS_NAME) == true) {
4505 2 : cloudsync_update_schema_hash(data);
4506 2 : }
4507 : }
4508 :
4509 4 : return DBRES_OK;
4510 4 : }
4511 :
4512 0 : int cloudsync_cleanup_all (cloudsync_context *data) {
4513 0 : return database_cleanup(data);
4514 : }
4515 :
4516 525 : int cloudsync_terminate (cloudsync_context *data) {
4517 : // can't use for/loop here because data->tables_count is changed by table_remove
4518 801 : while (data->tables_count > 0) {
4519 276 : cloudsync_table_context *t = data->tables[data->tables_count - 1];
4520 276 : table_remove(data, t);
4521 276 : table_free(t);
4522 : }
4523 :
4524 525 : if (data->schema_version_stmt) databasevm_finalize(data->schema_version_stmt);
4525 525 : if (data->data_version_stmt) databasevm_finalize(data->data_version_stmt);
4526 525 : if (data->db_version_stmt) databasevm_finalize(data->db_version_stmt);
4527 525 : if (data->getset_siteid_stmt) databasevm_finalize(data->getset_siteid_stmt);
4528 525 : if (data->current_schema) cloudsync_memory_free(data->current_schema);
4529 :
4530 525 : data->schema_version_stmt = NULL;
4531 525 : data->data_version_stmt = NULL;
4532 525 : data->db_version_stmt = NULL;
4533 525 : data->getset_siteid_stmt = NULL;
4534 525 : data->current_schema = NULL;
4535 :
4536 : // reset the site_id so the cloudsync_context_init will be executed again
4537 : // if any other cloudsync function is called after terminate
4538 525 : data->site_id[0] = 0;
4539 :
4540 525 : return 1;
4541 : }
4542 :
4543 307 : int cloudsync_init_table (cloudsync_context *data, const char *table_name, const char *algo_name, CLOUDSYNC_INIT_FLAG init_flags) {
4544 : // sanity check table and its primary key(s)
4545 307 : int rc = cloudsync_table_sanity_check(data, table_name, init_flags);
4546 307 : if (rc != DBRES_OK) return rc;
4547 :
4548 : // init cloudsync_settings
4549 305 : if (cloudsync_context_init(data) == NULL) {
4550 0 : return cloudsync_set_error(data, "Unable to initialize cloudsync context", DBRES_MISUSE);
4551 : }
4552 :
4553 : // sanity check algo name (if exists)
4554 305 : table_algo algo_new = table_algo_none;
4555 305 : if (!algo_name) algo_name = CLOUDSYNC_DEFAULT_ALGO;
4556 :
4557 305 : algo_new = cloudsync_algo_from_name(algo_name);
4558 305 : if (algo_new == table_algo_none) {
4559 : char buffer[1024];
4560 1 : snprintf(buffer, sizeof(buffer), "Unknown CRDT algorithm name %s", algo_name);
4561 1 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4562 : }
4563 :
4564 : // DWS and AWS algorithms are not yet implemented in the merge logic
4565 304 : if (algo_new == table_algo_crdt_dws || algo_new == table_algo_crdt_aws) {
4566 : char buffer[1024];
4567 2 : snprintf(buffer, sizeof(buffer), "CRDT algorithm %s is not yet supported", algo_name);
4568 2 : return cloudsync_set_error(data, buffer, DBRES_ERROR);
4569 : }
4570 :
4571 : // check if table name was already augmented
4572 302 : table_algo algo_current = dbutils_table_settings_get_algo(data, table_name);
4573 :
4574 : // sanity check algorithm
4575 302 : if ((algo_new == algo_current) && (algo_current != table_algo_none)) {
4576 : // if table algorithms and the same and not none, do nothing
4577 302 : } else if ((algo_new == table_algo_none) && (algo_current == table_algo_none)) {
4578 : // nothing is written into settings because the default table_algo_crdt_cls will be used
4579 0 : algo_new = algo_current = table_algo_crdt_cls;
4580 274 : } else if ((algo_new == table_algo_none) && (algo_current != table_algo_none)) {
4581 : // algo is already written into settins so just use it
4582 0 : algo_new = algo_current;
4583 274 : } else if ((algo_new != table_algo_none) && (algo_current == table_algo_none)) {
4584 : // write table algo name in settings
4585 274 : dbutils_table_settings_set_key_value(data, table_name, "*", "algo", algo_name);
4586 274 : } else {
4587 : // error condition
4588 0 : return cloudsync_set_error(data, "The function cloudsync_cleanup(table) must be called before changing a table algorithm", DBRES_MISUSE);
4589 : }
4590 :
4591 : // Run the following function even if table was already augmented.
4592 : // It is safe to call the following function multiple times, if there is nothing to update nothing will be changed.
4593 : // After an alter table, in contrast, all the cloudsync triggers, tables and stmts must be recreated.
4594 :
4595 : // sync algo with table (unused in this version)
4596 : // cloudsync_sync_table_key(data, table_name, "*", CLOUDSYNC_KEY_ALGO, crdt_algo_name(algo_new));
4597 :
4598 : // read row-level filter from settings (if any)
4599 : char init_filter_buf[2048];
4600 302 : int init_frc = dbutils_table_settings_get_value(data, table_name, "*", "filter", init_filter_buf, sizeof(init_filter_buf));
4601 302 : const char *init_filter = (init_frc == DBRES_OK && init_filter_buf[0]) ? init_filter_buf : NULL;
4602 :
4603 : // check triggers
4604 302 : rc = database_create_triggers(data, table_name, algo_new, init_filter);
4605 302 : if (rc != DBRES_OK) return cloudsync_set_error(data, "An error occurred while creating triggers", DBRES_MISUSE);
4606 :
4607 : // check meta-table
4608 302 : rc = database_create_metatable(data, table_name);
4609 302 : if (rc != DBRES_OK) return cloudsync_set_error(data, "An error occurred while creating metatable", DBRES_MISUSE);
4610 :
4611 : // add prepared statements
4612 302 : if (cloudsync_add_dbvms(data) != DBRES_OK) {
4613 0 : return cloudsync_set_error(data, "An error occurred while trying to compile prepared SQL statements", DBRES_MISUSE);
4614 : }
4615 :
4616 : // add table to in-memory data context
4617 302 : if (table_add_to_context(data, algo_new, table_name) == false) {
4618 : char buffer[1024];
4619 0 : snprintf(buffer, sizeof(buffer), "An error occurred while adding %s table information to global context", table_name);
4620 0 : return cloudsync_set_error(data, buffer, DBRES_MISUSE);
4621 : }
4622 :
4623 302 : if (cloudsync_refill_metatable(data, table_name) != DBRES_OK) {
4624 0 : return cloudsync_set_error(data, "An error occurred while trying to fill the augmented table", DBRES_MISUSE);
4625 : }
4626 :
4627 302 : return DBRES_OK;
4628 307 : }
|
