local DataStorage = require("datastorage")
local InfoMessage = require("ui/widget/infomessage")
local UIManager = require("ui/uimanager")
local SQ3 = require("lua-ljsqlite3/init")
local logger = require("logger")
local _ = require("readest_i18n")

local SyncStats = {}

local function db_path()
    return DataStorage:getSettingsDir() .. "/statistics.sqlite3"
end

-- Read book md5/title/authors + page events with start_time > cursor.
function SyncStats:collectSince(cursor)
    local conn = SQ3.open(db_path())
    local books, pages, seen = {}, {}, {}
    local stmt = conn:prepare([[
        SELECT b.md5, b.title, b.authors, p.page, p.start_time, p.duration, p.total_pages
        FROM page_stat_data p JOIN book b ON b.id = p.id_book
        WHERE p.start_time > ? ORDER BY p.start_time ASC]])
    stmt:reset():bind(tonumber(cursor) or 0)
    local row = stmt:step()
    while row ~= nil do
        local md5 = row[1]
        if md5 and not seen[md5] then
            seen[md5] = true
            table.insert(books, { book_hash = md5, title = row[2] or "", authors = row[3] or "" })
        end
        table.insert(pages, {
            book_hash = md5,
            page = tonumber(row[4]),
            start_time = tonumber(row[5]),
            duration = tonumber(row[6]),
            total_pages = tonumber(row[7]),
        })
        row = stmt:step()
    end
    stmt:close()
    conn:close()
    return books, pages
end

-- Fold duplicate book rows sharing one md5 into the row KOReader's native
-- statistics plugin actually reads. That plugin keys rows by (title,
-- authors, md5) — UNIQUE INDEX book_title_authors_md5 — so when its
-- extracted metadata drifts from Readest's, the first native open adds a
-- second, zeroed row and the synced reading time is stranded on the
-- sync-created one (#4861). Only rows the native plugin never adopted
-- (pages IS NULL; it always sets pages on the rows it creates or updates)
-- are folded, and an adopted row is never deleted: an open reader session
-- may hold its cached book id. live_book_id is the id the current session's
-- statistics module cached — a session that adopted a sync-created row keeps
-- pages NULL until its first close, so that row must not be folded either.
local function mergeDuplicateBooks(conn, touched, live_book_id)
    local dup_stmt = conn:prepare([[
        SELECT md5 FROM book WHERE md5 IS NOT NULL AND md5 != ''
        GROUP BY md5 HAVING COUNT(*) > 1;]])
    local dups = {}
    local row = dup_stmt:step()
    while row ~= nil do
        table.insert(dups, row[1])
        row = dup_stmt:step()
    end
    dup_stmt:close()
    if #dups == 0 then return end
    local survivor_stmt = conn:prepare([[
        SELECT id FROM book WHERE md5 = ?
        ORDER BY (pages IS NOT NULL) DESC, last_open DESC, id ASC LIMIT 1;]])
    local stranded_stmt = conn:prepare("SELECT id FROM book WHERE md5 = ? AND id != ? AND id != ? AND pages IS NULL;")
    local move_pages = conn:prepare([[
        INSERT INTO page_stat_data (id_book, page, start_time, duration, total_pages)
        SELECT ?, page, start_time, duration, total_pages
        FROM page_stat_data WHERE id_book = ?
        ON CONFLICT(id_book, page, start_time)
        DO UPDATE SET duration = max(duration, excluded.duration), total_pages = excluded.total_pages;]])
    local del_pages = conn:prepare("DELETE FROM page_stat_data WHERE id_book = ?;")
    local del_book = conn:prepare("DELETE FROM book WHERE id = ?;")
    for _, md5 in ipairs(dups) do
        local keep = tonumber(survivor_stmt:reset():bind(md5):step()[1])
        local stranded = {}
        stranded_stmt:reset():bind(md5, keep, tonumber(live_book_id) or -1)
        local r = stranded_stmt:step()
        while r ~= nil do
            table.insert(stranded, tonumber(r[1]))
            r = stranded_stmt:step()
        end
        for _, dead in ipairs(stranded) do
            move_pages:reset():bind(keep, dead):step()
            del_pages:reset():bind(dead):step()
            del_book:reset():bind(dead):step()
            touched[dead] = nil
            touched[keep] = true
            logger.dbg("ReadestStats applyRemote: folded duplicate book row "
                .. dead .. " into " .. keep .. " (md5=" .. md5 .. ")")
        end
    end
    survivor_stmt:close()
    stranded_stmt:close()
    move_pages:close()
    del_pages:close()
    del_book:close()
end

-- Upsert pulled rows into the local statistics.sqlite3 (union / longer-duration).
function SyncStats:applyRemote(books, pages, live_book_id)
    local conn = SQ3.open(db_path())
    conn:exec("BEGIN;")
    -- Insert a row only for an md5 this DB has never seen: if KOReader
    -- already tracks the book under its own (possibly drifted) metadata,
    -- adding one keyed on Readest's would duplicate it (#4861).
    local insert_book = conn:prepare([[
        INSERT INTO book (title, authors, md5)
        SELECT ?, ?, ? WHERE NOT EXISTS (SELECT 1 FROM book WHERE md5 = ?);]])
    for _, b in ipairs(books or {}) do
        insert_book:reset():bind(b.title or "", b.authors or "", b.book_hash, b.book_hash):step()
    end
    insert_book:close()
    -- Attach events to the row the native statistics plugin reads: it always
    -- sets pages and last_open on its rows; sync-created rows leave them NULL.
    local find_id = conn:prepare([[
        SELECT id FROM book WHERE md5 = ?
        ORDER BY (pages IS NOT NULL) DESC, last_open DESC, id ASC LIMIT 1;]])
    local insert_page = conn:prepare([[
        INSERT INTO page_stat_data (id_book, page, start_time, duration, total_pages)
        VALUES (?, ?, ?, ?, ?)
        ON CONFLICT(id_book, page, start_time)
        DO UPDATE SET duration = max(duration, excluded.duration), total_pages = excluded.total_pages;]])
    local id_cache = {}
    local touched = {}
    for _, p in ipairs(pages or {}) do
        local id = id_cache[p.book_hash]
        if not id then
            local r = find_id:reset():bind(p.book_hash):step()
            if r ~= nil then id = tonumber(r[1]); id_cache[p.book_hash] = id end
        end
        if id then
            insert_page:reset():bind(id, p.page, p.start_time, p.duration, p.total_pages):step()
            touched[id] = true
        end
    end
    find_id:close()
    insert_page:close()
    mergeDuplicateBooks(conn, touched, live_book_id)
    -- Mirror the Readest app's recomputeBookTotals so a KOReader device shows
    -- fresh totals right after a pull (id is a trusted integer from the DB).
    -- last_open never regresses: on native rows it is a real open timestamp
    -- that can be newer than the last synced reading event.
    for id in pairs(touched) do
        conn:exec(string.format([[
            UPDATE book SET
                total_read_time  = COALESCE((SELECT SUM(duration) FROM page_stat_data WHERE id_book = %d), 0),
                total_read_pages = COALESCE((SELECT COUNT(DISTINCT page) FROM page_stat_data WHERE id_book = %d), 0),
                last_open        = MAX(COALESCE(last_open, 0),
                                       COALESCE((SELECT MAX(start_time + duration) FROM page_stat_data WHERE id_book = %d), 0))
            WHERE id = %d;]], id, id, id, id))
    end
    conn:exec("COMMIT;")
    conn:close()
end

-- Page events per push request. A large backlog (e.g. a fresh push cursor
-- after pulling the full multi-device history, #5666) cannot go out as one
-- request: the sync client's 10s total socket timeout kills it, the cursor
-- never advances, and every retry re-sends the same payload — wedging "Push
-- stats now" forever. Mirrors PUSH_CHUNK in the app's statsSync.ts and the
-- server's stat_pages batch size.
local PUSH_CHUNK = 500

function SyncStats:push(settings, client, interactive)
    -- `settings` is the plain readest_sync data table (see main.lua:init), so
    -- the cursor is a field; persist by saving the whole table back to
    -- G_reader_settings, mirroring readest_syncauth.
    local cursor = settings.stats_push_cursor or 0
    local books, pages = self:collectSince(cursor)
    logger.dbg("ReadestStats push: cursor=" .. tostring(cursor)
        .. " collected books=" .. #books .. " pages=" .. #pages
        .. " interactive=" .. tostring(interactive))
    if #pages == 0 then
        logger.dbg("ReadestStats push: nothing to push (no page events past cursor)")
        if interactive then
            UIManager:show(InfoMessage:new{ text = _("Reading statistics are up to date"), timeout = 2 })
        end
        return
    end
    local book_by_hash = {}
    for _, b in ipairs(books) do book_by_hash[b.book_hash] = b end
    local function pushFrom(i)
        if i > #pages then
            logger.dbg("ReadestStats push: all " .. #pages .. " page(s) pushed")
            if interactive then
                UIManager:show(InfoMessage:new{ text = _("Reading statistics pushed"), timeout = 2 })
            end
            return
        end
        local last = math.min(i + PUSH_CHUNK - 1, #pages)
        -- Never split a start_time across chunks (pages are ordered by
        -- start_time): advancing the cursor past it would drop the remaining
        -- same-second events on resume.
        while last < #pages and pages[last + 1].start_time == pages[last].start_time do
            last = last + 1
        end
        local chunk_pages, chunk_books, seen = {}, {}, {}
        for j = i, last do
            local p = pages[j]
            table.insert(chunk_pages, p)
            local b = p.book_hash and book_by_hash[p.book_hash]
            if b and not seen[p.book_hash] then
                seen[p.book_hash] = true
                table.insert(chunk_books, b)
            end
        end
        logger.dbg("ReadestStats push: dispatching pages " .. i .. ".." .. last
            .. " of " .. #pages)
        -- pushChanges declares books/notes/configs as required_params (shared
        -- /sync POST contract, readest-sync-api.json); include them empty so
        -- Spore sends the request — the server defaults each to [] and
        -- processes statBooks/statPages independently
        -- (apps/readest-app/src/pages/api/sync.ts).
        client:pushChanges(
            { books = {}, notes = {}, configs = {}, statBooks = chunk_books, statPages = chunk_pages },
            function(success, body, status)
                logger.dbg("ReadestStats push: response success=" .. tostring(success)
                    .. " status=" .. tostring(status))
                if success then
                    settings.stats_push_cursor = pages[last].start_time
                    G_reader_settings:saveSetting("readest_sync", settings)
                    logger.dbg("ReadestStats push: cursor advanced to "
                        .. tostring(settings.stats_push_cursor))
                    -- nextTick so the next chunk starts from the event loop:
                    -- chaining inside the callback would nest a coroutine
                    -- resume per chunk on the synchronous (non-Turbo) HTTP
                    -- path and could blow the C stack on a huge backlog.
                    UIManager:nextTick(function() pushFrom(last + 1) end)
                else
                    logger.dbg("ReadestStats push: failed, cursor kept at "
                        .. tostring(settings.stats_push_cursor) .. "; body=" .. tostring(body))
                    if interactive then
                        UIManager:show(InfoMessage:new{ text = _("Failed to push reading statistics"), timeout = 2 })
                    end
                end
            end)
    end
    pushFrom(1)
end

-- Page events per pull request. The server pages stat_pages by updated_at and
-- completes the trailing millisecond of each page, so a strict
-- `since = cursor` next request never skips rows that share an updated_at
-- (same contract as PULL_PAGE in the app's statsSync.ts). A full multi-device
-- history would otherwise come back as one multi-MB response under the sync
-- client's 10s total socket timeout: the pull-side twin of #5666, failing
-- forever without ever advancing the cursor. Paging also keeps every
-- applyRemote transaction bounded on the device.
local PULL_PAGE = 1000

function SyncStats:pull(settings, client, interactive, logout_fn, ui)
    logger.dbg("ReadestStats pull: since=" .. tostring(settings.stats_pull_cursor or 0)
        .. " interactive=" .. tostring(interactive))
    local total_pages = 0
    local function pullFrom(since)
        -- pullChanges requires since/type/book/meta_hash params and accepts
        -- limit (readest-sync-api.json).
        client:pullChanges(
            { since = since, type = "stats", book = "", meta_hash = "", limit = PULL_PAGE },
            function(success, response, status)
                logger.dbg("ReadestStats pull: response success=" .. tostring(success)
                    .. " status=" .. tostring(status))
                if not success then
                    if status == 401 or status == 403 then
                        if logout_fn then logout_fn() end
                    end
                    -- Pages already applied stay applied and the cursor stays
                    -- at the last successful page, so a retry resumes there.
                    if interactive then
                        UIManager:show(InfoMessage:new{ text = _("Failed to pull reading statistics"), timeout = 2 })
                    end
                    return
                end
                local nbooks = response and response.statBooks and #response.statBooks or 0
                local npages = response and response.statPages and #response.statPages or 0
                logger.dbg("ReadestStats pull: applying statBooks=" .. nbooks .. " statPages=" .. npages
                    .. " since=" .. tostring(since))
                -- Resolved inside the callback so it reflects the session state
                -- at apply time, after the async network round-trip.
                local live_book_id = ui and ui.statistics and ui.statistics.id_curr_book or nil
                self:applyRemote(response.statBooks, response.statPages, live_book_id)
                total_pages = total_pages + npages
                local newest = since
                for _, p in ipairs(response.statPages or {}) do
                    local u = tonumber(p.updated_at_ms) or 0
                    if u > newest then newest = u end
                end
                if newest > since then
                    settings.stats_pull_cursor = newest
                    G_reader_settings:saveSetting("readest_sync", settings)
                    logger.dbg("ReadestStats pull: cursor advanced to " .. tostring(newest))
                else
                    logger.dbg("ReadestStats pull: cursor unchanged (no newer rows)")
                end
                -- A full page means more may follow; a short page (or no cursor
                -- progress) means we have caught up. nextTick so the next page
                -- starts from the event loop: chaining inside the callback would
                -- nest a coroutine resume per page on the synchronous (non-Turbo)
                -- HTTP path and could blow the C stack on a huge history.
                if npages >= PULL_PAGE and newest > since then
                    UIManager:nextTick(function() pullFrom(newest) end)
                    return
                end
                if interactive then
                    local text = total_pages > 0 and _("Reading statistics pulled")
                        or _("Reading statistics are up to date")
                    UIManager:show(InfoMessage:new{ text = text, timeout = 2 })
                end
            end)
    end
    pullFrom(settings.stats_pull_cursor or 0)
end

return SyncStats
