devs/cproof
1
2
Fork 1
Code Issues 43 Pull Requests 6 Actions Packages Projects 1 Releases Wiki Activity
Files
e91e8403b044886b10ae68bdd332e28cf6ef6e7b
cproof/src/database_flatfile.c
jabber.developer2 e91e8403b0 feat: complete field parity, harden export/import, add tests 
Export/Import improvements:
- Replace pagination with direct SQL query (db_sqlite_get_all_chat)
- Wrap import in SQL transaction with rollback on error
- Add fsync before fclose in export for data safety
- Sort merged output by timestamp with secondary key (stanza_id, from_jid)
- Export archive_id and marked_read from SQLite (lossless migration)
- Add progress indication every 500 messages during write/import
- Expand dedup key body prefix from 64 to 256 chars
- Fix g_slist_append O(n²) → g_slist_prepend + g_slist_reverse O(n)

Field parity (to_jid, to_resource, marked_read):
- Add fields to ff_parsed_line_t struct
- Write/parse to:|to_res:|read: metadata tags in flatfile format
- Pass to_resource through _ff_add_message and all callers
- Add marked_read to ProfMessage struct with -1 default (unset)
- Preserve fields across export/import round-trips

Tests (19 new: 11 unit + 8 functional):
- Unit: to_jid_and_marked_read, bracket_in_stanza_id, backslash_in_resource,
  mucpm_type, all_enc_types, crlf_handling, to_jid_special_chars,
  multiple_lines, parsed_line_free_null_safe, no_space_rejected,
  unclosed_bracket
- Functional: export_idempotent_no_duplicates, export_lmc_correction_survives,
  switch_preserves_old_backend_data, export_all_contacts,
  import_double_dedup, verify_after_export,
  switch_backends_independent_messages, export_empty_contact
- Rebalance test groups: move Chat Session from Group 3 to Group 4
  (25/33/30/27 instead of 25/33/36/21)
- Remove hardcoded test counts from group comments

Man page:
- Document /history switch sqlite|flatfile
2026-03-09 12:38:41 +03:00

998 lines
32 KiB
C
Raw Blame History

/*
* database_flatfile.c
* vim: expandtab:ts=4:sts=4:sw=4
*
* Copyright (C) 2026 Profanity Contributors
*
* This file is part of Profanity.
*
* Profanity is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Profanity is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with Profanity. If not, see <https://www.gnu.org/licenses/>.
*
* Flat-file database backend: init/close, message write, message read,
* LMC correction logic, and the backend vtable.
*
* Parser and escape helpers are in database_flatfile_parser.c.
* Integrity verification is in database_flatfile_verify.c.
* Shared types and prototypes are in database_flatfile.h.
*/
#include "config.h"
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/file.h>
#include <glib.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#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"
// =========================================================================
// Shared global — account JID stored during init for path construction
// =========================================================================
char* g_flatfile_account_jid = NULL;
static GHashTable* g_contact_states = NULL;
// =========================================================================
// Per-contact state / sparse index
// =========================================================================
ff_contact_state_t*
ff_state_new(const char* filepath)
{
ff_contact_state_t* state = g_malloc0(sizeof(ff_contact_state_t));
state->filepath = g_strdup(filepath);
state->cursor_offset = -1;
return state;
}
void
ff_state_free(ff_contact_state_t* state)
{
if (!state)
return;
g_free(state->filepath);
g_free(state->entries);
g_free(state);
}
static gboolean
_ff_state_build(ff_contact_state_t* state)
{
FILE* fp = fopen(state->filepath, "r");
if (!fp)
return FALSE;
int c1 = fgetc(fp);
int c2 = fgetc(fp);
int c3 = fgetc(fp);
if (c1 == 0xEF && c2 == 0xBB && c3 == 0xBF) {
state->bom_len = 3;
} else {
state->bom_len = 0;
fseek(fp, 0, SEEK_SET);
}
state->n_entries = 0;
state->total_lines = 0;
size_t msg_count = 0;
while (1) {
off_t pos = ftell(fp);
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;
}
state->total_lines++;
if (msg_count % FF_INDEX_STEP == 0) {
char* space = strchr(buf, ' ');
if (space) {
char* ts_str = g_strndup(buf, space - buf);
GDateTime* dt = g_date_time_new_from_iso8601(ts_str, NULL);
if (dt) {
if (state->n_entries >= state->cap_entries) {
state->cap_entries = state->cap_entries ? state->cap_entries * 2 : 64;
state->entries = g_realloc(state->entries,
state->cap_entries * sizeof(ff_index_entry_t));
}
state->entries[state->n_entries].byte_offset = pos;
state->entries[state->n_entries].timestamp_epoch = g_date_time_to_unix(dt);
state->n_entries++;
g_date_time_unref(dt);
}
g_free(ts_str);
}
}
msg_count++;
free(buf);
}
fclose(fp);
struct stat st;
if (stat(state->filepath, &st) == 0) {
state->stamp.mtime = st.st_mtime;
state->stamp.size = st.st_size;
state->stamp.inode = st.st_ino;
}
log_debug("flatfile: index built for %s (%zu entries, %zu lines)",
state->filepath, state->n_entries, state->total_lines);
return TRUE;
}
static gboolean
_ff_state_extend(ff_contact_state_t* state)
{
struct stat st;
if (stat(state->filepath, &st) != 0)
return FALSE;
if (st.st_size <= state->stamp.size)
return _ff_state_build(state);
FILE* fp = fopen(state->filepath, "r");
if (!fp)
return FALSE;
fseek(fp, state->stamp.size, SEEK_SET);
if (state->stamp.size > 0) {
fseek(fp, state->stamp.size - 1, SEEK_SET);
int prev = fgetc(fp);
if (prev != '\n' && prev != EOF) {
int ch;
while ((ch = fgetc(fp)) != EOF && ch != '\n') {
}
}
}
// Use a local counter for new messages so index step alignment
// doesn't depend on the total from the initial build.
size_t new_count = 0;
while (1) {
off_t pos = ftell(fp);
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;
}
state->total_lines++;
if (new_count % FF_INDEX_STEP == 0) {
char* space = strchr(buf, ' ');
if (space) {
char* ts_str = g_strndup(buf, space - buf);
GDateTime* dt = g_date_time_new_from_iso8601(ts_str, NULL);
if (dt) {
if (state->n_entries >= state->cap_entries) {
state->cap_entries = state->cap_entries ? state->cap_entries * 2 : 64;
state->entries = g_realloc(state->entries,
state->cap_entries * sizeof(ff_index_entry_t));
}
state->entries[state->n_entries].byte_offset = pos;
state->entries[state->n_entries].timestamp_epoch = g_date_time_to_unix(dt);
state->n_entries++;
g_date_time_unref(dt);
}
g_free(ts_str);
}
}
new_count++;
free(buf);
}
fclose(fp);
state->stamp.mtime = st.st_mtime;
state->stamp.size = st.st_size;
state->stamp.inode = st.st_ino;
log_debug("flatfile: index extended for %s (%zu entries, %zu lines)",
state->filepath, state->n_entries, state->total_lines);
return TRUE;
}
gboolean
ff_state_ensure_fresh(ff_contact_state_t* state)
{
if (!state || !state->filepath)
return FALSE;
struct stat st;
if (stat(state->filepath, &st) != 0) {
state->total_lines = 0;
state->n_entries = 0;
state->stamp.size = 0;
return FALSE;
}
if (state->stamp.size == 0 && state->n_entries == 0)
return _ff_state_build(state);
if (st.st_mtime == state->stamp.mtime
&& st.st_size == state->stamp.size
&& st.st_ino == state->stamp.inode) {
return TRUE;
}
if (st.st_ino == state->stamp.inode && st.st_size > state->stamp.size)
return _ff_state_extend(state);
state->cursor_offset = -1;
return _ff_state_build(state);
}
off_t
ff_state_offset_for_time(ff_contact_state_t* state, const char* iso_time)
{
if (!state || !iso_time || state->n_entries == 0)
return state ? (off_t)state->bom_len : 0;
GDateTime* dt = g_date_time_new_from_iso8601(iso_time, NULL);
if (!dt)
return state->bom_len;
gint64 target = g_date_time_to_unix(dt);
g_date_time_unref(dt);
size_t lo = 0, hi = state->n_entries;
while (lo < hi) {
size_t mid = lo + (hi - lo) / 2;
if (state->entries[mid].timestamp_epoch <= target)
lo = mid + 1;
else
hi = mid;
}
if (lo == 0)
return state->bom_len;
return state->entries[lo - 1].byte_offset;
}
static ff_contact_state_t*
_ff_get_state(const char* contact_barejid)
{
if (!g_contact_states || !contact_barejid)
return NULL;
ff_contact_state_t* state = g_hash_table_lookup(g_contact_states, contact_barejid);
if (state)
return state;
auto_gchar gchar* filepath = ff_get_log_path(contact_barejid);
if (!filepath)
return NULL;
state = ff_state_new(filepath);
g_hash_table_insert(g_contact_states, g_strdup(contact_barejid), state);
return state;
}
// =========================================================================
// Core write logic
// =========================================================================
static void
_ff_add_message(const char* type, const char* stanza_id, const char* archive_id,
const char* replace_id, const char* from_barejid, const char* from_resource,
const char* to_barejid, const char* to_resource, const char* message_text,
GDateTime* timestamp, prof_enc_t enc)
{
auto_gchar gchar* pref_dblog = prefs_get_string(PREF_DBLOG);
if (g_strcmp0(pref_dblog, "off") == 0) {
return;
}
// "redact" replaces the message content.
char* effective_msg = NULL;
if (g_strcmp0(pref_dblog, "redact") == 0) {
effective_msg = g_strdup("[REDACTED]");
} else {
effective_msg = g_strdup(message_text ? message_text : "");
}
GDateTime* dt = timestamp ? g_date_time_ref(timestamp) : g_date_time_new_now_local();
auto_gchar gchar* date_fmt = g_date_time_format_iso8601(dt);
g_date_time_unref(dt);
// Determine which JID to use for the log file path (the "other" party)
const char* contact = NULL;
const Jid* myjid = connection_get_jid();
if (myjid && myjid->barejid && g_strcmp0(from_barejid, myjid->barejid) == 0) {
contact = to_barejid;
} else {
contact = from_barejid;
}
auto_gchar gchar* log_path = ff_get_log_path(contact);
if (!log_path) {
log_error("flatfile: could not determine log path for %s", contact);
g_free(effective_msg);
return;
}
// Ensure directory exists
auto_gchar gchar* dir = g_path_get_dirname(log_path);
if (!ff_ensure_dir(dir)) {
g_free(effective_msg);
return;
}
// Open the file with O_NOFOLLOW to prevent symlink attacks,
// and O_APPEND for safe concurrent appends.
// Try O_CREAT|O_EXCL first to detect new files atomically (no TOCTOU).
int fd = open(log_path, O_WRONLY | O_APPEND | O_CREAT | O_EXCL | O_NOFOLLOW,
S_IRUSR | S_IWUSR);
gboolean is_new = (fd >= 0);
if (fd < 0) {
if (errno == EEXIST) {
// File already exists — open for append
fd = open(log_path, O_WRONLY | O_APPEND | O_NOFOLLOW);
}
if (fd < 0) {
// ELOOP (symlink) or other error
log_error("flatfile: could not open %s for writing (errno=%d: %s)",
log_path, errno, strerror(errno));
g_free(effective_msg);
return;
}
}
FILE* fp = fdopen(fd, "a");
if (!fp) {
log_error("flatfile: fdopen failed for %s (errno=%d)", log_path, errno);
close(fd);
g_free(effective_msg);
return;
}
// Advisory lock to prevent interleaved writes from concurrent instances
if (flock(fd, LOCK_EX) != 0) {
log_warning("flatfile: flock(%s) failed (errno=%d: %s), writing anyway",
log_path, errno, strerror(errno));
}
if (is_new) {
if (fprintf(fp, "%s", FLATFILE_HEADER) < 0) {
log_error("flatfile: failed to write header to %s (errno=%d)",
log_path, errno);
}
}
ff_write_line(fp, date_fmt, type, ff_get_message_enc_str(enc),
stanza_id, archive_id, replace_id,
from_barejid, from_resource,
to_barejid, to_resource, -1,
effective_msg);
fflush(fp);
// fclose also releases the flock
fclose(fp);
g_free(effective_msg);
}
// =========================================================================
// Backend callbacks: init / close
// =========================================================================
static gboolean
_flatfile_init(ProfAccount* account)
{
if (!account || !account->jid) {
log_error("flatfile: cannot init without account JID");
return FALSE;
}
g_free(g_flatfile_account_jid);
g_flatfile_account_jid = g_strdup(account->jid);
// Ensure base directory exists
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);
if (!ff_ensure_dir(base_dir)) {
return FALSE;
}
if (g_contact_states) {
g_hash_table_destroy(g_contact_states);
}
g_contact_states = g_hash_table_new_full(g_str_hash, g_str_equal,
g_free, (GDestroyNotify)ff_state_free);
log_info("Initialized flat-file database backend: %s", base_dir);
return TRUE;
}
static void
_flatfile_close(void)
{
if (g_contact_states) {
g_hash_table_destroy(g_contact_states);
g_contact_states = NULL;
}
g_free(g_flatfile_account_jid);
g_flatfile_account_jid = NULL;
log_debug("flatfile: closed");
}
// =========================================================================
// Backend callbacks: add message
// =========================================================================
// =========================================================================
// Incoming message validation helpers
// =========================================================================
// Search log file for a line containing aid:{archive_id}, return TRUE if found.
static gboolean
_ff_has_archive_id(const char* log_path, const char* archive_id)
{
if (!log_path || !archive_id || strlen(archive_id) == 0)
return FALSE;
FILE* fp = fopen(log_path, "r");
if (!fp)
return FALSE;
auto_gchar gchar* needle = g_strdup_printf("aid:%s", archive_id);
gboolean found = FALSE;
while (!found) {
gboolean trunc = FALSE;
char* line = ff_readline(fp, &trunc);
if (!line)
break;
if (trunc) {
free(line);
break;
}
if (strstr(line, needle)) {
found = TRUE;
}
free(line);
}
fclose(fp);
return found;
}
// Search log file for a line with stanza id matching replace_id, return the
// from_jid of the original message (caller must g_free). Returns NULL if not found.
static char*
_ff_find_original_sender(const char* log_path, const char* replace_id)
{
if (!log_path || !replace_id || strlen(replace_id) == 0)
return NULL;
FILE* fp = fopen(log_path, "r");
if (!fp)
return NULL;
auto_gchar gchar* needle = g_strdup_printf("id:%s", replace_id);
char* sender = NULL;
while (!sender) {
gboolean trunc = FALSE;
char* line = ff_readline(fp, &trunc);
if (!line)
break;
if (trunc) {
free(line);
break;
}
if (strstr(line, needle)) {
ff_parsed_line_t* pl = ff_parse_line(line);
if (pl && pl->stanza_id && g_strcmp0(pl->stanza_id, replace_id) == 0) {
sender = g_strdup(pl->from_jid);
}
if (pl)
ff_parsed_line_free(pl);
}
free(line);
}
fclose(fp);
return sender;
}
static void
_flatfile_add_incoming(ProfMessage* message)
{
if (!message || !message->from_jid)
return;
const Jid* to_jid = message->to_jid ? message->to_jid : connection_get_jid();
const char* type = ff_get_message_type_str(message->type);
// Determine contact JID for log path (same logic as _ff_add_message)
const char* contact = NULL;
const Jid* myjid = connection_get_jid();
if (myjid && myjid->barejid && g_strcmp0(message->from_jid->barejid, myjid->barejid) == 0) {
contact = to_jid->barejid;
} else {
contact = message->from_jid->barejid;
}
auto_gchar gchar* log_path = ff_get_log_path(contact);
// MAM dedup: skip if this stanza-id already exists in the log (XEP-0313 replay)
if (message->stanzaid && !message->is_mam && log_path) {
if (_ff_has_archive_id(log_path, message->stanzaid)) {
log_error("flatfile: duplicate stanza-id '%s' from %s, skipping",
message->stanzaid, message->from_jid->barejid);
cons_show_error("Got a message with duplicate (server-generated) stanza-id from %s.",
message->from_jid->fulljid);
return;
}
}
// LMC sender validation: verify the correction comes from the original sender
if (message->replace_id && log_path) {
auto_gchar gchar* original_sender = _ff_find_original_sender(log_path, message->replace_id);
if (original_sender && g_strcmp0(original_sender, message->from_jid->barejid) != 0) {
log_error("flatfile: LMC sender mismatch — corrected msg sender: %s, original: %s, replace-id: %s",
message->from_jid->barejid, original_sender, message->replace_id);
cons_show_error("%s sent a message correction with mismatched sender. See log for details.",
message->from_jid->barejid);
return;
}
}
_ff_add_message(type, message->id, message->stanzaid, message->replace_id,
message->from_jid->barejid, message->from_jid->resourcepart,
to_jid->barejid, to_jid->resourcepart, message->plain,
message->timestamp, message->enc);
}
static void
_flatfile_add_outgoing_chat(const char* const id, const char* const barejid,
const char* const message, const char* const replace_id, prof_enc_t enc)
{
const Jid* myjid = connection_get_jid();
_ff_add_message("chat", id, NULL, replace_id,
myjid ? myjid->barejid : "me", myjid ? myjid->resourcepart : NULL,
barejid, NULL, message, NULL, enc);
}
static void
_flatfile_add_outgoing_muc(const char* const id, const char* const barejid,
const char* const message, const char* const replace_id, prof_enc_t enc)
{
const Jid* myjid = connection_get_jid();
_ff_add_message("muc", id, NULL, replace_id,
myjid ? myjid->barejid : "me", myjid ? myjid->resourcepart : NULL,
barejid, NULL, message, NULL, enc);
}
static void
_flatfile_add_outgoing_muc_pm(const char* const id, const char* const barejid,
const char* const message, const char* const replace_id, prof_enc_t enc)
{
const Jid* myjid = connection_get_jid();
_ff_add_message("mucpm", id, NULL, replace_id,
myjid ? myjid->barejid : "me", myjid ? myjid->resourcepart : NULL,
barejid, NULL, message, NULL, enc);
}
// =========================================================================
// Read logic
// =========================================================================
// Apply LMC: for messages with corrects:X, find original and replace its text
static void
_ff_apply_lmc(GSList* parsed_lines, GSList** result)
{
// Build a hash: stanza_id -> parsed_line for quick lookup
GHashTable* id_map = g_hash_table_new(g_str_hash, g_str_equal);
for (GSList* l = parsed_lines; l; l = l->next) {
ff_parsed_line_t* pl = l->data;
if (pl->stanza_id && strlen(pl->stanza_id) > 0) {
g_hash_table_insert(id_map, pl->stanza_id, pl);
}
}
// Track which lines are corrections (skip them in output, apply to originals)
GHashTable* corrections = g_hash_table_new(g_direct_hash, g_direct_equal);
// Map: original line ptr -> latest correcting line ptr
GHashTable* correction_map = g_hash_table_new(g_direct_hash, g_direct_equal);
for (GSList* l = parsed_lines; l; l = l->next) {
ff_parsed_line_t* pl = l->data;
if (pl->replace_id && strlen(pl->replace_id) > 0) {
ff_parsed_line_t* original = g_hash_table_lookup(id_map, pl->replace_id);
if (original) {
// Follow chain to the root original, with cycle/depth guard
ff_parsed_line_t* root = original;
GHashTable* visited = g_hash_table_new(g_direct_hash, g_direct_equal);
g_hash_table_insert(visited, root, root);
int depth = 0;
while (root->replace_id && strlen(root->replace_id) > 0 && depth < FF_MAX_LMC_DEPTH) {
ff_parsed_line_t* parent = g_hash_table_lookup(id_map, root->replace_id);
if (parent && !g_hash_table_contains(visited, parent)) {
g_hash_table_insert(visited, parent, parent);
root = parent;
depth++;
} else {
break;
}
}
g_hash_table_destroy(visited);
if (depth >= FF_MAX_LMC_DEPTH) {
log_warning("flatfile: LMC correction chain too deep (>%d), ignoring", FF_MAX_LMC_DEPTH);
}
g_hash_table_insert(correction_map, root, pl);
g_hash_table_insert(corrections, pl, pl);
}
}
}
// Build result: for each non-correction line, output it (with corrected text if applicable)
for (GSList* l = parsed_lines; l; l = l->next) {
ff_parsed_line_t* pl = l->data;
if (g_hash_table_lookup(corrections, pl)) {
continue; // skip correction-only lines
}
ff_parsed_line_t* latest_correction = g_hash_table_lookup(correction_map, pl);
ProfMessage* msg;
if (latest_correction) {
// Use corrected text with original's metadata
msg = ff_parsed_to_profmessage(pl);
g_free(msg->plain);
msg->plain = g_strdup(latest_correction->message ? latest_correction->message : "");
} else {
msg = ff_parsed_to_profmessage(pl);
}
*result = g_slist_append(*result, msg);
}
g_hash_table_destroy(id_map);
g_hash_table_destroy(corrections);
g_hash_table_destroy(correction_map);
}
// =========================================================================
// Backend callbacks: query history
// =========================================================================
static db_history_result_t
_flatfile_get_previous_chat(const gchar* const contact_barejid, const gchar* start_time,
const gchar* end_time, gboolean from_start, gboolean flip,
GSList** result)
{
if (!g_flatfile_account_jid) {
log_warning("flatfile: get_previous_chat called but not initialized");
return DB_RESPONSE_ERROR;
}
const Jid* myjid = connection_get_jid();
if (!myjid || !myjid->barejid) {
log_warning("flatfile: no connection JID available");
return DB_RESPONSE_ERROR;
}
// Get or create per-contact state with sparse index
ff_contact_state_t* state = _ff_get_state(contact_barejid);
if (!state)
return DB_RESPONSE_ERROR;
// Ensure index is up-to-date (stat check -> build/extend if file changed)
if (!ff_state_ensure_fresh(state))
return DB_RESPONSE_EMPTY;
if (state->total_lines == 0)
return DB_RESPONSE_EMPTY;
// Reset cursor for filtered (non-Page-Up) queries
if (start_time)
state->cursor_offset = -1;
// Determine read byte-range using the sparse index
off_t read_from = state->bom_len;
off_t read_to = state->stamp.size;
// Use stored cursor for sequential Page Up (skip bisect)
if (!start_time && end_time && state->cursor_offset >= 0) {
read_to = state->cursor_offset;
} else if (end_time) {
// Upper-bound: find first index entry AFTER end_time
GDateTime* edt = g_date_time_new_from_iso8601(end_time, NULL);
if (edt) {
gint64 end_epoch = g_date_time_to_unix(edt);
g_date_time_unref(edt);
size_t lo = 0, hi = state->n_entries;
while (lo < hi) {
size_t mid = lo + (hi - lo) / 2;
if (state->entries[mid].timestamp_epoch <= end_epoch)
lo = mid + 1;
else
hi = mid;
}
// lo = first entry with timestamp > end_epoch
// Use next entry's offset as read_to (+ 1 entry margin)
if (lo + 1 < state->n_entries)
read_to = state->entries[lo + 1].byte_offset;
// else read_to stays at EOF
}
}
if (start_time) {
read_from = ff_state_offset_for_time(state, start_time);
}
// For "last N messages": back up from read_to using index
if (!from_start) {
int margin_entries = (MESSAGES_TO_RETRIEVE * 3) / FF_INDEX_STEP + 2;
// Find index entry closest to (but before) read_to
size_t entry_idx = 0;
for (size_t i = 0; i < state->n_entries; i++) {
if (state->entries[i].byte_offset >= read_to)
break;
entry_idx = i;
}
size_t start_entry = entry_idx > (size_t)margin_entries
? entry_idx - margin_entries
: 0;
off_t backed = state->entries[start_entry].byte_offset;
if (backed > read_from)
read_from = backed;
}
// Open file and read lines in [read_from, read_to)
FILE* fp = fopen(state->filepath, "r");
if (!fp)
return DB_RESPONSE_EMPTY;
fseek(fp, read_from, SEEK_SET);
// Pre-parse time filter boundaries once (avoid per-line allocation)
GDateTime* start_dt = start_time ? g_date_time_new_from_iso8601(start_time, NULL) : NULL;
GDateTime* end_dt = end_time ? g_date_time_new_from_iso8601(end_time, NULL) : NULL;
GSList* all_parsed = NULL;
while (1) {
off_t pos = ftell(fp);
if (pos < 0 || pos >= read_to)
break;
gboolean truncated = FALSE;
char* buf = ff_readline(fp, &truncated);
if (!buf)
break;
if (truncated) {
free(buf);
break;
}
ff_parsed_line_t* pl = ff_parse_line(buf);
free(buf);
if (!pl)
continue;
pl->file_offset = pos;
// Time filters
if (start_dt && g_date_time_compare(pl->timestamp, start_dt) <= 0) {
ff_parsed_line_free(pl);
continue;
}
if (end_dt && g_date_time_compare(pl->timestamp, end_dt) >= 0) {
ff_parsed_line_free(pl);
continue;
}
// JID filter
gboolean matches = FALSE;
if (myjid->barejid && contact_barejid) {
if ((g_strcmp0(pl->from_jid, contact_barejid) == 0)
|| (g_strcmp0(pl->from_jid, myjid->barejid) == 0)) {
matches = TRUE;
}
} else {
matches = TRUE;
}
if (!matches) {
ff_parsed_line_free(pl);
continue;
}
all_parsed = g_slist_append(all_parsed, pl);
}
fclose(fp);
if (start_dt)
g_date_time_unref(start_dt);
if (end_dt)
g_date_time_unref(end_dt);
if (!all_parsed)
return DB_RESPONSE_EMPTY;
// Update cursor: byte offset of oldest parsed message in this batch
ff_parsed_line_t* oldest = all_parsed->data;
state->cursor_offset = oldest->file_offset;
// Apply LMC corrections and build ProfMessage list
GSList* pre_result = NULL;
_ff_apply_lmc(all_parsed, &pre_result);
g_slist_free_full(all_parsed, (GDestroyNotify)ff_parsed_line_free);
if (!pre_result)
return DB_RESPONSE_EMPTY;
// Paginate: keep first or last MESSAGES_TO_RETRIEVE
guint total = g_slist_length(pre_result);
if (total > MESSAGES_TO_RETRIEVE) {
if (from_start) {
GSList* nth = g_slist_nth(pre_result, MESSAGES_TO_RETRIEVE);
if (nth) {
GSList* prev = g_slist_nth(pre_result, MESSAGES_TO_RETRIEVE - 1);
if (prev)
prev->next = NULL;
g_slist_free_full(nth, (GDestroyNotify)message_free);
}
} else {
guint skip = total - MESSAGES_TO_RETRIEVE;
GSList* keep_start = g_slist_nth(pre_result, skip);
GSList* prev = g_slist_nth(pre_result, skip - 1);
if (prev)
prev->next = NULL;
g_slist_free_full(pre_result, (GDestroyNotify)message_free);
pre_result = keep_start;
}
}
if (flip) {
pre_result = g_slist_reverse(pre_result);
}
*result = pre_result;
return g_slist_length(*result) != 0 ? DB_RESPONSE_SUCCESS : DB_RESPONSE_EMPTY;
}
static ProfMessage*
_flatfile_get_limits_info(const gchar* const contact_barejid, gboolean is_last)
{
ProfMessage* msg = message_init();
if (!g_flatfile_account_jid || !contact_barejid) {
if (is_last)
msg->timestamp = g_date_time_new_now_utc();
return msg;
}
ff_contact_state_t* state = _ff_get_state(contact_barejid);
if (!state) {
if (is_last)
msg->timestamp = g_date_time_new_now_utc();
return msg;
}
if (!ff_state_ensure_fresh(state) || state->total_lines == 0) {
if (is_last)
msg->timestamp = g_date_time_new_now_utc();
return msg;
}
FILE* fp = fopen(state->filepath, "r");
if (!fp) {
if (is_last)
msg->timestamp = g_date_time_new_now_utc();
return msg;
}
ff_parsed_line_t* found = NULL;
if (!is_last) {
// First message: start from beginning (skip BOM)
fseek(fp, state->bom_len, SEEK_SET);
char* buf;
while ((buf = ff_readline(fp, NULL)) != NULL) {
found = ff_parse_line(buf);
free(buf);
if (found)
break;
}
} else {
// Last message: seek near end using index for efficiency
off_t seek_pos = 0;
if (state->n_entries > 0)
seek_pos = state->entries[state->n_entries - 1].byte_offset;
fseek(fp, seek_pos, SEEK_SET);
char* buf;
while ((buf = ff_readline(fp, NULL)) != NULL) {
ff_parsed_line_t* pl = ff_parse_line(buf);
free(buf);
if (pl) {
if (found)
ff_parsed_line_free(found);
found = pl;
}
}
}
fclose(fp);
if (found) {
msg->stanzaid = found->archive_id ? g_strdup(found->archive_id) : NULL;
msg->timestamp = g_date_time_ref(found->timestamp);
ff_parsed_line_free(found);
} else if (is_last) {
msg->timestamp = g_date_time_new_now_utc();
}
return msg;
}
// =========================================================================
// Backend vtable
// =========================================================================
static db_backend_t flatfile_backend = {
.name = "flatfile",
.init = _flatfile_init,
.close = _flatfile_close,
.add_incoming = _flatfile_add_incoming,
.add_outgoing_chat = _flatfile_add_outgoing_chat,
.add_outgoing_muc = _flatfile_add_outgoing_muc,
.add_outgoing_muc_pm = _flatfile_add_outgoing_muc_pm,
.get_previous_chat = _flatfile_get_previous_chat,
.get_limits_info = _flatfile_get_limits_info,
.verify_integrity = ff_verify_integrity,
};
db_backend_t*
db_backend_flatfile(void)
{
return &flatfile_backend;
}
Reference in New Issue View Git Blame Copy Permalink