Files
cproof/src/database_export.c
jabber.developer2 7ba7e53291 refactor: extract helpers for verify print, scan loop, and per-contact export
- extract _show_integrity_issues() from cmd_history verify block
- extract _ff_scan_lines() shared by _ff_state_build and _ff_state_extend
- extract _export_one_contact() from log_database_export_to_flatfile loop
2026-03-28 12:58:27 +03:00

582 lines
20 KiB
C

// SPDX-License-Identifier: GPL-3.0-or-later
//
// This file is part of CProof.
// See LICENSE for the full GPLv3 text and the special OpenSSL linking exception.
/*
* database_export.c
* vim: expandtab:ts=4:sts=4:sw=4
*
* Cross-backend export/import for message history.
* Reads from one backend, writes to the other, with merge + dedup.
*/
#include "config.h"
#include <glib.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/file.h>
#ifdef HAVE_SQLITE
#include <sqlite3.h>
#endif
#include "log.h"
#include "common.h"
#include "config/files.h"
#include "database.h"
#include "database_flatfile.h"
#include "config/preferences.h"
#include "ui/ui.h"
#include "xmpp/xmpp.h"
#include "xmpp/message.h"
// =========================================================================
// Everything below is only used when HAVE_SQLITE is defined
// =========================================================================
#ifdef HAVE_SQLITE
// Print progress to console every N messages during export/import
#define EXPORT_PROGRESS_INTERVAL 500
// =========================================================================
// Dedup key: stanza_id if present, else hash of timestamp+from+body
// =========================================================================
static char*
_make_dedup_key(const char* stanza_id, const char* timestamp, const char* from_jid, const char* body)
{
if (stanza_id && strlen(stanza_id) > 0)
return g_strdup(stanza_id);
// Fallback: composite key from timestamp + sender + body prefix
auto_gchar gchar* raw = g_strdup_printf("%s|%s|%.256s",
timestamp ? timestamp : "",
from_jid ? from_jid : "",
body ? body : "");
return g_compute_checksum_for_string(G_CHECKSUM_SHA256, raw, -1);
}
// =========================================================================
// Helper: reverse ff_jid_to_dir (best-effort)
// '_at_' -> '@'
// =========================================================================
static char*
_dir_to_jid(const char* dirname)
{
if (!dirname)
return NULL;
return str_replace(dirname, "_at_", "@");
}
// =========================================================================
// Helper: list all contact JIDs from flatfile directory
// Returns a GSList of g_strdup'd JID strings. Caller frees with g_slist_free_full(list, g_free).
// =========================================================================
static GSList*
_ff_list_contacts(void)
{
GSList* contacts = NULL;
if (!g_flatfile_account_jid)
return NULL;
auto_gchar gchar* data_path = files_get_data_path(DIR_FLATLOG);
auto_gchar gchar* my_dir = ff_jid_to_dir(g_flatfile_account_jid);
auto_gchar gchar* base_dir = g_strdup_printf("%s/%s", data_path, my_dir);
GError* err = NULL;
GDir* dir = g_dir_open(base_dir, 0, &err);
if (!dir) {
if (err) {
log_warning("export: cannot open flatlog dir %s: %s", base_dir, err->message);
g_error_free(err);
} else {
log_warning("export: cannot open flatlog dir %s (no error details)", base_dir);
}
return NULL;
}
const gchar* entry;
while ((entry = g_dir_read_name(dir)) != NULL) {
auto_gchar gchar* full_path = g_strdup_printf("%s/%s/history.log", base_dir, entry);
if (g_file_test(full_path, G_FILE_TEST_IS_REGULAR)) {
char* jid = _dir_to_jid(entry);
if (jid) {
contacts = g_slist_prepend(contacts, jid);
}
}
}
g_dir_close(dir);
return g_slist_reverse(contacts);
}
// =========================================================================
// Helper: read ALL parsed lines from a flatfile for a contact
// Returns GSList<ff_parsed_line_t*>. Caller frees with g_slist_free_full(list, (GDestroyNotify)ff_parsed_line_free).
// =========================================================================
static GSList*
_ff_read_all_lines(const char* contact_barejid)
{
GSList* lines = NULL;
auto_gchar gchar* log_path = ff_get_log_path(contact_barejid);
if (!log_path)
return NULL;
FILE* fp = fopen(log_path, "r");
if (!fp)
return NULL;
while (1) {
gboolean truncated = FALSE;
char* buf = ff_readline(fp, &truncated);
if (!buf)
break;
if (truncated) {
free(buf);
break;
}
if (buf[0] == '\0' || buf[0] == '#') {
free(buf);
continue;
}
ff_parsed_line_t* pl = ff_parse_line(buf);
free(buf);
if (pl) {
lines = g_slist_prepend(lines, pl);
}
}
fclose(fp);
return g_slist_reverse(lines);
}
// =========================================================================
// Comparator for sorting ff_parsed_line_t by timestamp (ISO8601 is lexicographic)
// =========================================================================
static gint
_compare_parsed_lines_by_timestamp(gconstpointer a, gconstpointer b)
{
const ff_parsed_line_t* la = a;
const ff_parsed_line_t* lb = b;
int cmp = g_strcmp0(la->timestamp_str, lb->timestamp_str);
if (cmp != 0)
return cmp;
// Secondary key: stanza_id (stabilises sort for same-second messages)
cmp = g_strcmp0(la->stanza_id, lb->stanza_id);
if (cmp != 0)
return cmp;
return g_strcmp0(la->from_jid, lb->from_jid);
}
// =========================================================================
// Export: SQLite -> flatfile (merge)
// =========================================================================
// Callback data for the SQLite query
typedef struct
{
GHashTable* seen_keys; // dedup set
FILE* fp; // output file
int exported; // counter
int skipped; // counter
} export_ctx_t;
// Export a single contact's messages from SQLite to flatfile.
// Returns the number of newly exported messages, or -1 on error/skip.
static int
_export_one_contact(const char* contact)
{
// 1. Read existing flatfile lines for dedup
GHashTable* seen_keys = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
GSList* existing = _ff_read_all_lines(contact);
for (GSList* l = existing; l; l = l->next) {
ff_parsed_line_t* pl = l->data;
char* key = _make_dedup_key(pl->stanza_id, pl->timestamp_str, pl->from_jid, pl->message);
g_hash_table_add(seen_keys, key);
}
int existing_count = g_slist_length(existing);
// 2. Query SQLite for this contact — get ALL messages via direct SQL
GSList* sqlite_lines = db_sqlite_get_all_chat(contact);
if (!sqlite_lines) {
log_debug("export: no SQLite messages for %s", contact);
g_hash_table_destroy(seen_keys);
g_slist_free_full(existing, (GDestroyNotify)ff_parsed_line_free);
return -1;
}
// 3. Merge: write existing flatfile lines + new SQLite lines to temp file
auto_gchar gchar* log_path = ff_get_log_path(contact);
if (!log_path) {
g_hash_table_destroy(seen_keys);
g_slist_free_full(existing, (GDestroyNotify)ff_parsed_line_free);
g_slist_free_full(sqlite_lines, (GDestroyNotify)message_free);
return -1;
}
// Ensure directory
auto_gchar gchar* dir = g_path_get_dirname(log_path);
ff_ensure_dir(dir);
auto_gchar gchar* tmp_path = g_strdup_printf("%s.export.tmp", log_path);
// Use open() with O_NOFOLLOW|O_EXCL to prevent symlink attacks,
// and explicit 0600 permissions (chat history is private).
int tmp_fd = open(tmp_path, O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW, S_IRUSR | S_IWUSR);
if (tmp_fd < 0) {
if (errno == EEXIST) {
// Stale temp file from previous failed export — remove and retry
unlink(tmp_path);
tmp_fd = open(tmp_path, O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW, S_IRUSR | S_IWUSR);
}
}
if (tmp_fd < 0) {
log_error("export: cannot create %s: %s", tmp_path, strerror(errno));
g_hash_table_destroy(seen_keys);
g_slist_free_full(existing, (GDestroyNotify)ff_parsed_line_free);
g_slist_free_full(sqlite_lines, (GDestroyNotify)message_free);
return -1;
}
FILE* fp = fdopen(tmp_fd, "w");
if (!fp) {
log_error("export: fdopen failed for %s: %s", tmp_path, strerror(errno));
close(tmp_fd);
unlink(tmp_path);
g_hash_table_destroy(seen_keys);
g_slist_free_full(existing, (GDestroyNotify)ff_parsed_line_free);
g_slist_free_full(sqlite_lines, (GDestroyNotify)message_free);
return -1;
}
// Lock the temp file
int fd = fileno(fp);
if (flock(fd, LOCK_EX) != 0) {
log_warning("export: flock(%s) failed: %s", tmp_path, strerror(errno));
}
fprintf(fp, "%s", FLATFILE_HEADER);
int contact_exported = 0;
int contact_skipped = 0;
// Collect ALL lines into a merged list for sorted output
GSList* merged = NULL;
// Add existing flatfile lines (mark as already seen in dedup)
for (GSList* l = existing; l; l = l->next) {
ff_parsed_line_t* pl = l->data;
// Create a shallow copy for the merged list (originals freed separately)
ff_parsed_line_t* copy = g_malloc0(sizeof(ff_parsed_line_t));
copy->timestamp_str = g_strdup(pl->timestamp_str);
copy->timestamp = pl->timestamp ? g_date_time_ref(pl->timestamp) : NULL;
copy->type = g_strdup(pl->type);
copy->enc = g_strdup(pl->enc);
copy->stanza_id = g_strdup(pl->stanza_id);
copy->archive_id = g_strdup(pl->archive_id);
copy->replace_id = g_strdup(pl->replace_id);
copy->from_jid = g_strdup(pl->from_jid);
copy->from_resource = g_strdup(pl->from_resource);
copy->to_jid = g_strdup(pl->to_jid);
copy->to_resource = g_strdup(pl->to_resource);
copy->marked_read = pl->marked_read;
copy->message = g_strdup(pl->message);
merged = g_slist_prepend(merged, copy);
}
// Add SQLite messages that aren't already in the flatfile
for (GSList* l = sqlite_lines; l; l = l->next) {
ProfMessage* msg = l->data;
if (!msg || !msg->timestamp)
continue;
auto_gchar gchar* ts = g_date_time_format_iso8601(msg->timestamp);
const char* from_jid = msg->from_jid ? msg->from_jid->barejid : "unknown";
const char* body = msg->plain ? msg->plain : "";
const char* sid = msg->id ? msg->id : "";
char* key = _make_dedup_key(sid, ts, from_jid, body);
if (g_hash_table_contains(seen_keys, key)) {
g_free(key);
contact_skipped++;
continue;
}
g_hash_table_add(seen_keys, key);
ff_parsed_line_t* pl = g_malloc0(sizeof(ff_parsed_line_t));
pl->timestamp_str = g_strdup(ts);
pl->timestamp = g_date_time_ref(msg->timestamp);
pl->type = g_strdup(ff_get_message_type_str(msg->type));
pl->enc = g_strdup(ff_get_message_enc_str(msg->enc));
pl->stanza_id = g_strdup(sid);
pl->archive_id = msg->stanzaid ? g_strdup(msg->stanzaid) : NULL;
pl->replace_id = NULL;
pl->from_jid = g_strdup(from_jid);
pl->from_resource = msg->from_jid ? g_strdup(msg->from_jid->resourcepart) : NULL;
pl->to_jid = msg->to_jid ? g_strdup(msg->to_jid->barejid) : NULL;
pl->to_resource = msg->to_jid ? g_strdup(msg->to_jid->resourcepart) : NULL;
pl->marked_read = msg->marked_read;
pl->message = g_strdup(body);
merged = g_slist_prepend(merged, pl);
contact_exported++;
}
// Sort merged list by timestamp (ISO8601 is lexicographically sortable)
merged = g_slist_sort(merged, _compare_parsed_lines_by_timestamp);
// Write all lines sorted
int written = 0;
for (GSList* l = merged; l; l = l->next) {
ff_parsed_line_t* pl = l->data;
ff_write_line(fp, pl->timestamp_str,
pl->type, pl->enc,
pl->stanza_id, pl->archive_id, pl->replace_id,
pl->from_jid, pl->from_resource,
pl->to_jid, pl->to_resource, pl->marked_read,
pl->message);
written++;
if (written % EXPORT_PROGRESS_INTERVAL == 0) {
cons_show(" ... %s: %d/%d written so far", contact, written, (int)g_slist_length(merged));
}
}
g_slist_free_full(merged, (GDestroyNotify)ff_parsed_line_free);
fflush(fp);
fsync(fd); // ensure data is on disk before atomic rename
fclose(fp); // also releases flock
// Atomic rename
int result = -1;
if (rename(tmp_path, log_path) != 0) {
log_error("export: rename %s -> %s failed: %s", tmp_path, log_path, strerror(errno));
cons_show_error("Export failed for %s: could not rename temp file (%s)", contact, strerror(errno));
unlink(tmp_path);
} else {
cons_show("Exported %s: %d new, %d skipped (already existed: %d)",
contact, contact_exported, contact_skipped, existing_count);
result = contact_exported;
}
g_hash_table_destroy(seen_keys);
g_slist_free_full(existing, (GDestroyNotify)ff_parsed_line_free);
g_slist_free_full(sqlite_lines, (GDestroyNotify)message_free);
return result;
}
int
log_database_export_to_flatfile(const gchar* const contact_jid)
{
const Jid* myjid = connection_get_jid();
if (!myjid || !myjid->barejid) {
cons_show_error("Export failed: not connected.");
return -1;
}
if (!g_flatfile_account_jid) {
// Flatfile backend not initialized — init it temporarily
// We need g_flatfile_account_jid for path construction
g_flatfile_account_jid = g_strdup(myjid->barejid);
}
// Get the SQLite backend
db_backend_t* sqlite_be = db_backend_sqlite();
if (!sqlite_be) {
cons_show_error("Export failed: SQLite backend not available.");
return -1;
}
// Determine contact list
GSList* contacts = NULL;
if (contact_jid) {
contacts = g_slist_append(contacts, g_strdup(contact_jid));
} else {
contacts = db_sqlite_list_contacts();
if (!contacts) {
cons_show("No contacts found in SQLite database.");
return 0;
}
cons_show("Exporting %d contact(s) to flat-file format...", g_slist_length(contacts));
}
int total_exported = 0;
for (GSList* c = contacts; c; c = c->next) {
const char* contact = c->data;
int exported = _export_one_contact(contact);
if (exported > 0)
total_exported += exported;
}
g_slist_free_full(contacts, g_free);
return total_exported;
}
// =========================================================================
// Import: flatfile -> SQLite (merge)
// =========================================================================
int
log_database_import_from_flatfile(const gchar* const contact_jid)
{
const Jid* myjid = connection_get_jid();
if (!myjid || !myjid->barejid) {
cons_show_error("Import failed: not connected.");
return -1;
}
if (!g_flatfile_account_jid) {
g_flatfile_account_jid = g_strdup(myjid->barejid);
}
db_backend_t* sqlite_be = db_backend_sqlite();
if (!sqlite_be) {
cons_show_error("Import failed: SQLite backend not available.");
return -1;
}
// Determine contacts
GSList* contacts = NULL;
if (contact_jid) {
contacts = g_slist_append(contacts, g_strdup(contact_jid));
} else {
contacts = _ff_list_contacts();
if (!contacts) {
cons_show("No flat-file history found to import.");
return 0;
}
}
int total_imported = 0;
for (GSList* c = contacts; c; c = c->next) {
const char* contact = c->data;
// 1. Read all flatfile lines
GSList* ff_lines = _ff_read_all_lines(contact);
if (!ff_lines) {
log_debug("import: no flatfile messages for %s", contact);
continue;
}
// 2. Build dedup set from SQLite (existing stanza_ids + fallback keys)
GHashTable* seen_keys = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
// Read all SQLite messages for this contact via direct SQL query
GSList* existing_msgs = db_sqlite_get_all_chat(contact);
for (GSList* l = existing_msgs; l; l = l->next) {
ProfMessage* msg = l->data;
if (!msg)
continue;
auto_gchar gchar* ts = msg->timestamp ? g_date_time_format_iso8601(msg->timestamp) : g_strdup("");
char* key = _make_dedup_key(msg->id, ts, msg->from_jid ? msg->from_jid->barejid : "", msg->plain);
g_hash_table_add(seen_keys, key);
}
g_slist_free_full(existing_msgs, (GDestroyNotify)message_free);
// 3. For each flatfile line not in seen_keys, insert via add_incoming
// Wrap in a transaction for atomicity and performance
int contact_imported = 0;
int contact_skipped = 0;
int total_lines = g_slist_length(ff_lines);
db_sqlite_begin_transaction();
gboolean import_ok = TRUE;
for (GSList* l = ff_lines; l; l = l->next) {
ff_parsed_line_t* pl = l->data;
if (!pl || !pl->timestamp)
continue;
auto_gchar gchar* ts = g_date_time_format_iso8601(pl->timestamp);
char* key = _make_dedup_key(pl->stanza_id, ts, pl->from_jid, pl->message);
if (g_hash_table_contains(seen_keys, key)) {
g_free(key);
contact_skipped++;
continue;
}
g_hash_table_add(seen_keys, key);
// Build a ProfMessage and insert via add_incoming
// (add_incoming uses _add_to_db which preserves the original timestamp)
gboolean is_outgoing = (g_strcmp0(pl->from_jid, myjid->barejid) == 0);
ProfMessage* msg = message_init();
msg->id = pl->stanza_id ? g_strdup(pl->stanza_id) : NULL;
msg->stanzaid = pl->archive_id ? g_strdup(pl->archive_id) : NULL;
msg->plain = g_strdup(pl->message ? pl->message : "");
msg->timestamp = g_date_time_ref(pl->timestamp);
msg->type = ff_get_message_type_type(pl->type);
msg->enc = ff_get_message_enc_type(pl->enc);
msg->replace_id = pl->replace_id ? g_strdup(pl->replace_id) : NULL;
msg->is_mam = TRUE; // Suppress dedup check in the backend
if (is_outgoing) {
msg->from_jid = jid_create(myjid->barejid);
if (pl->to_jid) {
msg->to_jid = jid_create_from_bare_and_resource(pl->to_jid, pl->to_resource);
} else {
msg->to_jid = jid_create(contact);
}
} else {
msg->from_jid = jid_create_from_bare_and_resource(pl->from_jid, pl->from_resource);
if (pl->to_jid) {
msg->to_jid = jid_create_from_bare_and_resource(pl->to_jid, pl->to_resource);
} else {
msg->to_jid = jid_create(myjid->barejid);
}
}
sqlite_be->add_incoming(msg);
message_free(msg);
// Verify the row was actually inserted
if (db_sqlite_last_changes() < 1) {
log_error("import: insert failed for %s at %s", contact, ts);
import_ok = FALSE;
break;
}
contact_imported++;
if (contact_imported % EXPORT_PROGRESS_INTERVAL == 0) {
cons_show(" ... %s: %d/%d imported so far", contact, contact_imported, total_lines);
}
}
if (import_ok) {
db_sqlite_end_transaction();
} else {
db_sqlite_rollback_transaction();
cons_show_error("Import of %s failed — transaction rolled back.", contact);
}
cons_show("Imported %s: %d new, %d skipped (already in SQLite)",
contact, contact_imported, contact_skipped);
total_imported += contact_imported;
g_hash_table_destroy(seen_keys);
g_slist_free_full(ff_lines, (GDestroyNotify)ff_parsed_line_free);
}
g_slist_free_full(contacts, g_free);
return total_imported;
}
#endif /* HAVE_SQLITE */