devs/cproof
1
2
Fork 1
Code Issues 43 Pull Requests 6 Actions Packages Projects 1 Releases Wiki Activity

database_flatfile: single-file storage with sparse index

Browse Source 
Replace per-contact directory structure with a single flat file per contact.
Each file uses pipe-delimited records with a sparse byte-offset index
that is rebuilt on demand and cached in memory.

- Rewrite _ff_write_record / _ff_read_records for single-file format
- Add sparse index (every Nth record) for O(log N) seek on history read
- Cursor-based pagination: _ff_get_previous_chat uses saved byte offset
- LMC corrections applied in-place during read (_ff_apply_lmc)
- Newline escaping (\n) in message bodies for line-based storage
- Simplify verify: parse + timestamp order + LMC reference check
- Update parser helpers for new field layout
This commit is contained in:
jabber.developer2
2026-02-19 16:09:23 +03:00
parent d25f90899c
commit 16ae7fb85c
4 changed files with 608 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files

578
src/database_flatfile.c
View File

@@ -31,6 +31,8 @@
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <glib.h>
#include <stdio.h>
#include <stdlib.h>
@@ -52,6 +54,261 @@
// =========================================================================
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') {
}
}
}
size_t msg_count = state->total_lines;
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);
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
@@ -79,6 +336,7 @@ _ff_add_message(const char* type, const char* stanza_id, const char* archive_id,
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;
@@ -89,8 +347,7 @@ _ff_add_message(const char* type, const char* stanza_id, const char* archive_id,
contact = from_barejid;
}
auto_gchar gchar* log_path = ff_get_log_path(contact, dt);
g_date_time_unref(dt);
auto_gchar gchar* log_path = ff_get_log_path(contact);
if (!log_path) {
log_error("flatfile: could not determine log path for %s", contact);
@@ -170,6 +427,12 @@ _flatfile_init(ProfAccount* account)
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;
}
@@ -177,6 +440,10 @@ _flatfile_init(ProfAccount* account)
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");
@@ -235,113 +502,6 @@ _flatfile_add_outgoing_muc_pm(const char* const id, const char* const barejid,
// Read logic
// =========================================================================
// List all log files for a contact, sorted by filename (date order)
static GSList*
_ff_list_log_files(const char* contact_barejid)
{
auto_gchar gchar* contact_dir = ff_get_contact_dir(contact_barejid);
if (!contact_dir || !g_file_test(contact_dir, G_FILE_TEST_IS_DIR)) {
return NULL;
}
GDir* dir = g_dir_open(contact_dir, 0, NULL);
if (!dir)
return NULL;
GSList* files = NULL;
const gchar* fname;
while ((fname = g_dir_read_name(dir)) != NULL) {
if (g_str_has_suffix(fname, ".log")) {
files = g_slist_insert_sorted(files,
g_strdup_printf("%s/%s", contact_dir, fname),
(GCompareFunc)g_strcmp0);
}
}
g_dir_close(dir);
return files;
}
// Read all parsed lines from a file, applying filters
static GSList*
_ff_read_file_messages(const char* filepath, const char* contact_barejid,
const char* start_time, const char* end_time,
const Jid* myjid)
{
GSList* messages = NULL;
FILE* fp = fopen(filepath, "r");
if (!fp)
return NULL;
// BOM detection
int c1 = fgetc(fp);
int c2 = fgetc(fp);
int c3 = fgetc(fp);
if (!(c1 == 0xEF && c2 == 0xBB && c3 == 0xBF)) {
// Not a BOM, rewind
fseek(fp, 0, SEEK_SET);
}
char* buf = NULL;
gboolean truncated = FALSE;
while ((buf = ff_readline(fp, &truncated)) != NULL) {
if (truncated) {
log_warning("flatfile: truncated line at EOF in %s, skipping", filepath);
free(buf);
break;
}
ff_parsed_line_t* pl = ff_parse_line(buf);
free(buf);
if (!pl)
continue;
// Check timestamp bounds
if (start_time) {
GDateTime* start_dt = g_date_time_new_from_iso8601(start_time, NULL);
if (start_dt) {
if (g_date_time_compare(pl->timestamp, start_dt) <= 0) {
g_date_time_unref(start_dt);
ff_parsed_line_free(pl);
continue;
}
g_date_time_unref(start_dt);
}
}
if (end_time) {
GDateTime* end_dt = g_date_time_new_from_iso8601(end_time, NULL);
if (end_dt) {
if (g_date_time_compare(pl->timestamp, end_dt) >= 0) {
g_date_time_unref(end_dt);
ff_parsed_line_free(pl);
continue;
}
g_date_time_unref(end_dt);
}
}
// Check JID filter: message must involve contact and our JID
gboolean matches = FALSE;
if (myjid && 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; // no filter
}
if (!matches) {
ff_parsed_line_free(pl);
continue;
}
messages = g_slist_append(messages, pl);
}
fclose(fp);
return messages;
}
// Apply LMC: for messages with corrects:X, find original and replace its text
static void
_ff_apply_lmc(GSList* parsed_lines, GSList** result)
@@ -435,52 +595,160 @@ _flatfile_get_previous_chat(const gchar* const contact_barejid, const gchar* sta
return DB_RESPONSE_ERROR;
}
// If no end_time, use now
auto_gchar gchar* effective_end = NULL;
if (end_time) {
effective_end = g_strdup(end_time);
} else {
GDateTime* now = g_date_time_new_now_local();
effective_end = g_date_time_format_iso8601(now);
g_date_time_unref(now);
}
// 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;
// Get all log files for this contact
GSList* files = _ff_list_log_files(contact_barejid);
if (!files) {
// 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
}
}
// Collect all matching parsed lines across files
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;
for (GSList* f = files; f; f = f->next) {
GSList* file_msgs = _ff_read_file_messages(f->data, contact_barejid,
start_time, effective_end, myjid);
all_parsed = g_slist_concat(all_parsed, file_msgs);
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);
g_slist_free_full(files, g_free);
if (start_dt)
g_date_time_unref(start_dt);
if (end_dt)
g_date_time_unref(end_dt);
if (!all_parsed) {
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);
// Free parsed lines
g_slist_free_full(all_parsed, (GDestroyNotify)ff_parsed_line_free);
if (!pre_result) {
if (!pre_result)
return DB_RESPONSE_EMPTY;
}
// Apply pagination: take first or last MESSAGES_TO_RETRIEVE
// Paginate: keep first or last MESSAGES_TO_RETRIEVE
guint total = g_slist_length(pre_result);
if (total > MESSAGES_TO_RETRIEVE) {
if (from_start) {
// Keep first N, free rest
GSList* nth = g_slist_nth(pre_result, MESSAGES_TO_RETRIEVE);
if (nth) {
GSList* prev = g_slist_nth(pre_result, MESSAGES_TO_RETRIEVE - 1);
@@ -489,7 +757,6 @@ _flatfile_get_previous_chat(const gchar* const contact_barejid, const gchar* sta
g_slist_free_full(nth, (GDestroyNotify)message_free);
}
} else {
// Keep last N, free first (total - N)
guint skip = total - MESSAGES_TO_RETRIEVE;
GSList* keep_start = g_slist_nth(pre_result, skip);
GSList* prev = g_slist_nth(pre_result, skip - 1);
@@ -500,7 +767,6 @@ _flatfile_get_previous_chat(const gchar* const contact_barejid, const gchar* sta
}
}
// Apply flip (reverse order)
if (flip) {
pre_result = g_slist_reverse(pre_result);
}
@@ -520,39 +786,32 @@ _flatfile_get_limits_info(const gchar* const contact_barejid, gboolean is_last)
return msg;
}
GSList* files = _ff_list_log_files(contact_barejid);
if (!files) {
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;
}
// For first message: read first non-comment line of first file
// For last message: read last non-comment line of last file
const char* target_file = is_last ? (const char*)g_slist_last(files)->data
: (const char*)files->data;
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(target_file, "r");
FILE* fp = fopen(state->filepath, "r");
if (!fp) {
g_slist_free_full(files, g_free);
if (is_last)
msg->timestamp = g_date_time_new_now_utc();
return msg;
}
// BOM skip
int c1 = fgetc(fp);
int c2 = fgetc(fp);
int c3 = fgetc(fp);
if (!(c1 == 0xEF && c2 == 0xBB && c3 == 0xBF)) {
fseek(fp, 0, SEEK_SET);
}
char* buf = NULL;
ff_parsed_line_t* found = NULL;
if (!is_last) {
// Find first valid line
// 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);
@@ -560,7 +819,13 @@ _flatfile_get_limits_info(const gchar* const contact_barejid, gboolean is_last)
break;
}
} else {
// Find last valid line — read all and keep last
// 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);
@@ -581,7 +846,6 @@ _flatfile_get_limits_info(const gchar* const contact_barejid, gboolean is_last)
msg->timestamp = g_date_time_new_now_utc();
}
g_slist_free_full(files, g_free);
return msg;
}