profanity/src/database_flatfile.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_flatfile.c
 * vim: expandtab:ts=4:sts=4:sw=4
 *
 * 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->archive_ids = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
    state->stanza_senders = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, g_free);
    return state;
}

void
ff_state_free(ff_contact_state_t* state)
{
    if (!state)
        return;
    g_free(state->filepath);
    g_free(state->entries);
    if (state->archive_ids)
        g_hash_table_destroy(state->archive_ids);
    if (state->stanza_senders)
        g_hash_table_destroy(state->stanza_senders);
    g_free(state);
}

// Lightweight ID extraction from a raw log line for the dedup/LMC cache.
// Avoids full ff_parse_line() overhead (no GDateTime, no message unescape).
static void
_ff_cache_line_ids(ff_contact_state_t* state, const char* line)
{
    // Find metadata section: first '[' to first unescaped ']'
    const char* bracket = strchr(line, '[');
    if (!bracket)
        return;

    const char* close = ff_find_unescaped_char(bracket + 1, ']');
    if (!close)
        return;

    // Split metadata on unescaped '|'
    char* meta_str = g_strndup(bracket + 1, close - bracket - 1);
    char** parts = ff_split_meta(meta_str);
    g_free(meta_str);

    char* stanza_id = NULL;
    char* archive_id = NULL;

    if (parts) {
        for (int i = 0; parts[i]; i++) {
            if (g_str_has_prefix(parts[i], FF_META_PREFIX_ID) && !stanza_id) {
                stanza_id = ff_unescape_meta_value(parts[i] + strlen(FF_META_PREFIX_ID));
            } else if (g_str_has_prefix(parts[i], FF_META_PREFIX_AID) && !archive_id) {
                archive_id = ff_unescape_meta_value(parts[i] + strlen(FF_META_PREFIX_AID));
            }
        }
        g_strfreev(parts);
    }

    // Extract from_jid (needed for stanza_senders map): after "] " before "/" or ": "
    char* from_jid = NULL;
    if (stanza_id && close[1] != '\0' && close[1] == ' ' && close[2] != '\0') {
        const char* sender_start = close + 2;
        const char* colonspace = ff_find_unescaped_colonspace(sender_start);
        if (colonspace) {
            const char* slash = memchr(sender_start, '/', colonspace - sender_start);
            const char* jid_end = slash ? slash : colonspace;
            from_jid = g_strndup(sender_start, jid_end - sender_start);
        }
    }

    if (archive_id && archive_id[0] != '\0') {
        g_hash_table_add(state->archive_ids, archive_id);
    } else {
        g_free(archive_id);
    }

    if (stanza_id && stanza_id[0] != '\0' && from_jid) {
        g_hash_table_insert(state->stanza_senders, stanza_id, from_jid);
    } else {
        g_free(stanza_id);
        g_free(from_jid);
    }
}

// Attempt to add an index entry for the current line.
// Called every FF_INDEX_STEP messages during build/extend.
static void
_ff_maybe_index_line(ff_contact_state_t* state, const char* buf, off_t pos)
{
    char* space = strchr(buf, ' ');
    if (!space)
        return;

    char* ts_str = g_strndup(buf, space - buf);
    GDateTime* dt = g_date_time_new_from_iso8601(ts_str, NULL);
    if (!dt) {
        log_warning("flatfile: unparsable timestamp in %s at offset %ld: %s",
                    state->filepath, (long)pos, ts_str);
        g_free(ts_str);
        return;
    }
    g_free(ts_str);

    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);
}

// Scan lines from an already-positioned file pointer, caching IDs and
// building index entries. Returns the number of messages scanned.
static size_t
_ff_scan_lines(ff_contact_state_t* state, FILE* fp)
{
    size_t 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++;
        _ff_cache_line_ids(state, buf);

        if (count % FF_INDEX_STEP == 0) {
            _ff_maybe_index_line(state, buf, pos);
        }
        count++;
        free(buf);
    }
    return count;
}

static gboolean
_ff_state_build(ff_contact_state_t* state)
{
    FILE* fp = fopen(state->filepath, "r");
    if (!fp)
        return FALSE;

    state->bom_len = ff_skip_bom(fp);

    int file_version = ff_read_format_version(fp);
    if (file_version == 0) {
        log_warning("flatfile: %s has no format-version marker, expected %d",
                    state->filepath, FLATFILE_FORMAT_VERSION);
    } else if (file_version != FLATFILE_FORMAT_VERSION) {
        log_warning("flatfile: %s format-version %d does not match expected %d",
                    state->filepath, file_version, FLATFILE_FORMAT_VERSION);
    }

    state->n_entries = 0;
    state->total_lines = 0;

    // Clear ID caches for full rebuild
    g_hash_table_remove_all(state->archive_ids);
    g_hash_table_remove_all(state->stanza_senders);

    _ff_scan_lines(state, fp);

    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 _ff_scan_lines for the shared read/index/cache loop.
    _ff_scan_lines(state, fp);

    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);

    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 Jid* myjid = connection_get_jid();
    gboolean is_outgoing = myjid && myjid->barejid
                           && g_strcmp0(from_barejid, myjid->barejid) == 0;
    const char* contact = is_outgoing ? to_barejid : 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.
    // Try O_CREAT|O_EXCL first to detect new files atomically (no TOCTOU).
    // O_APPEND is included for consistency — it has no effect on a new file
    // but ensures append semantics if the fd is inherited after the fallback.
    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 (O(1) via ID cache)
// =========================================================================

// Check if archive_id already exists in the contact's log via cached set.
static gboolean
_ff_has_archive_id(const char* contact_barejid, const char* archive_id)
{
    if (!contact_barejid || !archive_id || archive_id[0] == '\0')
        return FALSE;

    ff_contact_state_t* state = _ff_get_state(contact_barejid);
    if (!state)
        return FALSE;

    ff_state_ensure_fresh(state);
    return g_hash_table_contains(state->archive_ids, archive_id);
}

// Find the from_jid of the original message with given stanza_id via cached map.
// Caller must g_free the result.
static char*
_ff_find_original_sender(const char* contact_barejid, const char* replace_id)
{
    if (!contact_barejid || !replace_id || replace_id[0] == '\0')
        return NULL;

    ff_contact_state_t* state = _ff_get_state(contact_barejid);
    if (!state)
        return NULL;

    ff_state_ensure_fresh(state);
    const char* sender = g_hash_table_lookup(state->stanza_senders, replace_id);
    return sender ? g_strdup(sender) : NULL;
}

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;
    }
    // Stanza-id duplicate warning (non-blocking).
    // XEP-0359 stanza-ids SHOULD be unique per server, but older clients
    // (Pidgin, Adium, old Profanity) may use incremental IDs that servers
    // echo back, causing false positives.  Log but do NOT skip — matches
    // SQLite backend behavior.  See https://github.com/profanity-im/profanity/issues/1899
    if (message->stanzaid && !message->is_mam) {
        if (_ff_has_archive_id(contact, message->stanzaid)) {
            log_warning("flatfile: duplicate stanza-id '%s' from %s (may be non-unique client ID)",
                        message->stanzaid, message->from_jid->barejid);
        }
    }

    // LMC sender validation: verify the correction comes from the original sender
    if (message->replace_id) {
        auto_gchar gchar* original_sender = _ff_find_original_sender(contact, 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_prepend(*result, msg);
    }

    *result = g_slist_reverse(*result);

    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;
    // n_entries can be 0 even when total_lines > 0 if every line in the
    // file failed _ff_maybe_index_line (e.g. unparsable timestamps).
    // Subsequent backup logic dereferences entries[0] unconditionally,
    // which would segfault on a NULL entries array.
    if (state->n_entries == 0)
        return DB_RESPONSE_EMPTY;

    // Determine read byte-range using the sparse index
    off_t read_from = state->bom_len;
    off_t read_to = state->stamp.size;

    // Upper-bound via bisect on end_time. The previous cursor-based
    // shortcut (use last batch's oldest file_offset as read_to) was
    // both buggy and unnecessary: cursor was set to the oldest line in
    // the read window, but pagination dropped the older portion of that
    // window before returning to the caller. The next page-up therefore
    // started reading from a point much earlier than what the user had
    // actually seen, "jumping" past entire ranges of messages. Bisect
    // on the index is O(log n_entries) — for typical 200-entry indexes
    // it's a handful of comparisons, well below the per-line parse
    // cost — so we now run it unconditionally.
    if (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;
    }

    // Defensive: an empty or inverted range means there is nothing to read.
    // Reachable when start_time bisects past end_time, or when an end_time
    // upper-bound shrinks read_to below the BOM-adjusted read_from.
    if (read_from >= read_to) {
        log_debug("flatfile: empty range [%ld, %ld) for %s",
                  (long)read_from, (long)read_to, contact_barejid);
        return DB_RESPONSE_EMPTY;
    }

    // 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_prepend(all_parsed, pl);
    }
    fclose(fp);

    all_parsed = g_slist_reverse(all_parsed);

    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;

    // Apply LMC corrections (if enabled) and build ProfMessage list
    GSList* pre_result = NULL;
    if (prefs_get_boolean(PREF_CORRECTION_ALLOW)) {
        _ff_apply_lmc(all_parsed, &pre_result);
    } else {
        // LMC disabled — convert all lines to ProfMessage without correction
        for (GSList* cur = all_parsed; cur; cur = cur->next) {
            ff_parsed_line_t* pl = cur->data;
            if (pl) {
                ProfMessage* msg = ff_parsed_to_profmessage(pl);
                if (msg)
                    pre_result = g_slist_prepend(pre_result, msg);
            }
        }
        pre_result = g_slist_reverse(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;
}