package shine.db.dao;

import shine.db.DbController;
import shine.db.entities.DmDeliveryStateEntry;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/** Хранилище изменяемого состояния межсерверной доставки DM-пары. */
public final class DmDeliveryStateDAO {
    private static volatile DmDeliveryStateDAO instance;
    private final DbController db = DbController.getInstance();

    private DmDeliveryStateDAO() {}

    public static DmDeliveryStateDAO getInstance() {
        if (instance == null) {
            synchronized (DmDeliveryStateDAO.class) {
                if (instance == null) instance = new DmDeliveryStateDAO();
            }
        }
        return instance;
    }

    public DmDeliveryStateEntry upsertPair(
            String outgoingMessageKey,
            String eventId,
            String baseKey,
            String fromLogin,
            String toLogin,
            String incomingMessageKey,
            long createdAtMs,
            long expiresAtMs,
            int initialState,
            String deliveredServerLogin,
            String routesHash,
            boolean assistImmediately
    ) throws SQLException {
        long now = System.currentTimeMillis();
        int safeState = normalizeState(initialState, deliveredServerLogin);
        if (expiresAtMs <= now && safeState == DmDeliveryStateEntry.PENDING_NONE) {
            safeState = DmDeliveryStateEntry.FAILED_FINAL;
        }
        Long nextAttemptAt = safeState == DmDeliveryStateEntry.ACCEPTED
                ? now
                : null;
        String safeDeliveredLogin = safeState == DmDeliveryStateEntry.DELIVERED_ONE
                ? normalize(deliveredServerLogin)
                : null;

        try (Connection c = db.getConnection()) {
            String sql = """
                INSERT INTO dm_delivery_state (
                    outgoing_message_key, event_id, base_key, from_login, to_login,
                    incoming_message_key, created_at_ms, delivery_expires_at_ms,
                    delivery_state, delivered_server_login, recipient_routes_hash,
                    attempt_index, next_attempt_at_ms, last_attempt_at_ms, last_error, updated_at_ms
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, NULL, NULL, ?)
                ON CONFLICT (outgoing_message_key) DO UPDATE SET
                    event_id = EXCLUDED.event_id,
                    base_key = EXCLUDED.base_key,
                    from_login = EXCLUDED.from_login,
                    to_login = EXCLUDED.to_login,
                    incoming_message_key = EXCLUDED.incoming_message_key,
                    created_at_ms = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.created_at_ms
                        ELSE EXCLUDED.created_at_ms
                    END,
                    delivery_expires_at_ms = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.delivery_expires_at_ms
                        ELSE EXCLUDED.delivery_expires_at_ms
                    END,
                    delivery_state = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.delivery_state
                        ELSE EXCLUDED.delivery_state
                    END,
                    delivered_server_login = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.delivered_server_login
                        ELSE EXCLUDED.delivered_server_login
                    END,
                    recipient_routes_hash = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN COALESCE(dm_delivery_state.recipient_routes_hash, EXCLUDED.recipient_routes_hash)
                        ELSE EXCLUDED.recipient_routes_hash
                    END,
                    attempt_index = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.attempt_index
                        ELSE 0
                    END,
                    next_attempt_at_ms = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.next_attempt_at_ms
                        ELSE EXCLUDED.next_attempt_at_ms
                    END,
                    last_attempt_at_ms = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.last_attempt_at_ms
                        ELSE NULL
                    END,
                    last_error = CASE
                        WHEN dm_delivery_state.event_id = EXCLUDED.event_id THEN dm_delivery_state.last_error
                        ELSE NULL
                    END,
                    updated_at_ms = EXCLUDED.updated_at_ms
                """;
            try (PreparedStatement ps = c.prepareStatement(sql)) {
                ps.setString(1, outgoingMessageKey);
                ps.setString(2, eventId);
                ps.setString(3, baseKey);
                ps.setString(4, normalize(fromLogin));
                ps.setString(5, normalize(toLogin));
                ps.setString(6, incomingMessageKey);
                ps.setLong(7, createdAtMs);
                ps.setLong(8, expiresAtMs);
                ps.setInt(9, safeState);
                setNullableString(ps, 10, safeDeliveredLogin);
                setNullableString(ps, 11, normalizeBlank(routesHash));
                setNullableLong(ps, 12, nextAttemptAt);
                ps.setLong(13, now);
                ps.executeUpdate();
            }
        }
        DmDeliveryStateEntry result;
        if (initialState != DmDeliveryStateEntry.PENDING_NONE || !isBlank(deliveredServerLogin)) {
            result = mergeRemote(eventId, initialState, deliveredServerLogin, routesHash, now);
        } else {
            result = getByEventId(eventId);
        }
        if (result != null && result.getDeliveryExpiresAtMs() <= now
                && result.getDeliveryState() == DmDeliveryStateEntry.PENDING_NONE) {
            return finishAtExpiry(eventId, now);
        }
        return result;
    }

    public DmDeliveryStateEntry getByEventId(String eventId) throws SQLException {
        if (isBlank(eventId)) return null;
        try (Connection c = db.getConnection();
             PreparedStatement ps = c.prepareStatement(selectColumns() + " WHERE event_id = ?")) {
            ps.setString(1, eventId.trim());
            try (ResultSet rs = ps.executeQuery()) {
                return rs.next() ? mapRow(rs) : null;
            }
        }
    }

    public DmDeliveryStateEntry getByOutgoingMessageKey(String outgoingMessageKey) throws SQLException {
        if (isBlank(outgoingMessageKey)) return null;
        try (Connection c = db.getConnection();
             PreparedStatement ps = c.prepareStatement(selectColumns() + " WHERE outgoing_message_key = ?")) {
            ps.setString(1, outgoingMessageKey.trim());
            try (ResultSet rs = ps.executeQuery()) {
                return rs.next() ? mapRow(rs) : null;
            }
        }
    }

    public Map<String, DmDeliveryStateEntry> listByOutgoingMessageKeys(List<String> keys) throws SQLException {
        Map<String, DmDeliveryStateEntry> out = new HashMap<>();
        if (keys == null || keys.isEmpty()) return out;
        String placeholders = String.join(",", java.util.Collections.nCopies(keys.size(), "?"));
        String sql = selectColumns() + " WHERE outgoing_message_key IN (" + placeholders + ")";
        try (Connection c = db.getConnection(); PreparedStatement ps = c.prepareStatement(sql)) {
            for (int i = 0; i < keys.size(); i++) ps.setString(i + 1, keys.get(i));
            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) {
                    DmDeliveryStateEntry row = mapRow(rs);
                    out.put(row.getOutgoingMessageKey(), row);
                }
            }
        }
        return out;
    }

    public List<DmDeliveryStateEntry> listDue(long nowMs, int limit) throws SQLException {
        String sql = selectColumns() + """
             WHERE delivery_state = 0
               AND next_attempt_at_ms IS NOT NULL
               AND next_attempt_at_ms <= ?
             ORDER BY next_attempt_at_ms ASC, created_at_ms ASC
             LIMIT ?
            """;
        List<DmDeliveryStateEntry> out = new ArrayList<>();
        try (Connection c = db.getConnection(); PreparedStatement ps = c.prepareStatement(sql)) {
            ps.setLong(1, nowMs);
            ps.setInt(2, Math.max(1, limit));
            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) out.add(mapRow(rs));
            }
        }
        return out;
    }

    public boolean claimAttempt(DmDeliveryStateEntry row, long nowMs, Long nextAttemptAtMs) throws SQLException {
        String sql = """
            UPDATE dm_delivery_state
            SET attempt_index = attempt_index + 1,
                next_attempt_at_ms = ?,
                last_attempt_at_ms = ?,
                last_error = NULL,
                updated_at_ms = ?
            WHERE outgoing_message_key = ?
              AND event_id = ?
              AND attempt_index = ?
              AND delivery_state = 0
              AND next_attempt_at_ms IS NOT NULL
              AND next_attempt_at_ms <= ?
            """;
        try (Connection c = db.getConnection(); PreparedStatement ps = c.prepareStatement(sql)) {
            long networkLeaseUntilMs = nowMs + 60_000L;
            Long claimedNextAttemptAtMs = nextAttemptAtMs == null
                    ? networkLeaseUntilMs
                    : Math.max(nextAttemptAtMs, networkLeaseUntilMs);
            setNullableLong(ps, 1, claimedNextAttemptAtMs);
            ps.setLong(2, nowMs);
            ps.setLong(3, nowMs);
            ps.setString(4, row.getOutgoingMessageKey());
            ps.setString(5, row.getEventId());
            ps.setInt(6, row.getAttemptIndex());
            ps.setLong(7, nowMs);
            return ps.executeUpdate() == 1;
        }
    }

    public DmDeliveryStateEntry mergeRemote(
            String eventId,
            int remoteState,
            String remoteDeliveredLogin,
            String remoteRoutesHash,
            long nowMs
    ) throws SQLException {
        try (Connection c = db.getConnection()) {
            boolean previousAutoCommit = c.getAutoCommit();
            c.setAutoCommit(false);
            try {
                DmDeliveryStateEntry current = getByEventIdForUpdate(c, eventId);
                if (current == null) {
                    c.rollback();
                    return null;
                }
                MergeResult merged = merge(
                        current.getDeliveryState(),
                        current.getDeliveredServerLogin(),
                        current.getRecipientRoutesHash(),
                        remoteState,
                        remoteDeliveredLogin,
                        remoteRoutesHash
                );
                updateMutableState(c, current, merged.state(), merged.deliveredLogin(),
                        merged.routesHash(), current.getNextAttemptAtMs(), current.getLastError(), nowMs);
                c.commit();
                return getByEventId(eventId);
            } catch (Exception e) {
                try { c.rollback(); } catch (Exception ignored) {}
                throw e;
            } finally {
                c.setAutoCommit(previousAutoCommit);
            }
        }
    }

    public DmDeliveryStateEntry updateAfterAttempt(
            String eventId,
            int attemptedState,
            String deliveredServerLogin,
            String routesHash,
            Long nextAttemptAtMs,
            String lastError,
            long nowMs
    ) throws SQLException {
        try (Connection c = db.getConnection()) {
            boolean previousAutoCommit = c.getAutoCommit();
            c.setAutoCommit(false);
            try {
                DmDeliveryStateEntry current = getByEventIdForUpdate(c, eventId);
                if (current == null) {
                    c.rollback();
                    return null;
                }
                MergeResult merged = merge(
                        current.getDeliveryState(), current.getDeliveredServerLogin(), current.getRecipientRoutesHash(),
                        attemptedState, deliveredServerLogin, routesHash
                );
                Long effectiveNext = merged.state() == DmDeliveryStateEntry.ACCEPTED
                        ? nextAttemptAtMs
                        : null;
                updateMutableState(c, current, merged.state(), merged.deliveredLogin(),
                        merged.routesHash(), effectiveNext, lastError, nowMs);
                c.commit();
                return getByEventId(eventId);
            } catch (Exception e) {
                try { c.rollback(); } catch (Exception ignored) {}
                throw e;
            } finally {
                c.setAutoCommit(previousAutoCommit);
            }
        }
    }

    public DmDeliveryStateEntry finishAtExpiry(String eventId, long nowMs) throws SQLException {
        try (Connection c = db.getConnection()) {
            String sql = """
                UPDATE dm_delivery_state
                SET delivery_state = CASE WHEN delivery_state = 0 THEN 3 ELSE delivery_state END,
                    delivered_server_login = CASE WHEN delivery_state = 0 THEN NULL ELSE delivered_server_login END,
                    next_attempt_at_ms = NULL,
                    last_error = CASE WHEN delivery_state = 0 THEN 'DELIVERY_EXPIRED' ELSE last_error END,
                    updated_at_ms = ?
                WHERE event_id = ? AND delivery_state = 0
                """;
            try (PreparedStatement ps = c.prepareStatement(sql)) {
                ps.setLong(1, nowMs);
                ps.setString(2, eventId);
                ps.executeUpdate();
            }
        }
        return getByEventId(eventId);
    }

    /** Read-only peer confirmed that at least one recipient server accepted the message. */
    public DmDeliveryStateEntry markDeliveredFromPeer(String eventId, long nowMs) throws SQLException {
        try (Connection c = db.getConnection();
             PreparedStatement ps = c.prepareStatement("""
                 UPDATE dm_delivery_state
                 SET delivery_state = 1,
                     delivered_server_login = NULL,
                     next_attempt_at_ms = NULL,
                     last_error = NULL,
                     updated_at_ms = ?
                 WHERE event_id = ? AND delivery_state = 0
                 """)) {
            ps.setLong(1, nowMs);
            ps.setString(2, eventId);
            ps.executeUpdate();
        }
        return getByEventId(eventId);
    }

    public int removeByBaseKey(String baseKey) throws SQLException {
        try (Connection c = db.getConnection();
             PreparedStatement ps = c.prepareStatement("DELETE FROM dm_delivery_state WHERE base_key = ?")) {
            ps.setString(1, baseKey);
            return ps.executeUpdate();
        }
    }

    public int removeConversationBefore(String fromLogin, String toLogin, long boundaryTimeMs) throws SQLException {
        String sql = """
            DELETE FROM dm_delivery_state
            WHERE created_at_ms < ?
              AND ((LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
                OR (LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?)))
            """;
        try (Connection c = db.getConnection(); PreparedStatement ps = c.prepareStatement(sql)) {
            ps.setLong(1, boundaryTimeMs);
            ps.setString(2, fromLogin);
            ps.setString(3, toLogin);
            ps.setString(4, toLogin);
            ps.setString(5, fromLogin);
            return ps.executeUpdate();
        }
    }

    public int removeMissingMessages() throws SQLException {
        String sql = """
            DELETE FROM dm_delivery_state d
            WHERE NOT EXISTS (
                SELECT 1 FROM signed_messages m WHERE m.message_key = d.outgoing_message_key
            )
            """;
        try (Connection c = db.getConnection(); PreparedStatement ps = c.prepareStatement(sql)) {
            return ps.executeUpdate();
        }
    }

    private DmDeliveryStateEntry getByEventIdForUpdate(Connection c, String eventId) throws SQLException {
        try (PreparedStatement ps = c.prepareStatement(selectColumns() + " WHERE event_id = ? FOR UPDATE")) {
            ps.setString(1, eventId);
            try (ResultSet rs = ps.executeQuery()) {
                return rs.next() ? mapRow(rs) : null;
            }
        }
    }

    private void updateMutableState(
            Connection c,
            DmDeliveryStateEntry current,
            int state,
            String deliveredLogin,
            String routesHash,
            Long nextAttemptAtMs,
            String lastError,
            long nowMs
    ) throws SQLException {
        String sql = """
            UPDATE dm_delivery_state
            SET delivery_state = ?,
                delivered_server_login = ?,
                recipient_routes_hash = ?,
                next_attempt_at_ms = ?,
                last_error = ?,
                updated_at_ms = ?
            WHERE outgoing_message_key = ? AND event_id = ?
            """;
        try (PreparedStatement ps = c.prepareStatement(sql)) {
            ps.setInt(1, state);
            setNullableString(ps, 2, state == DmDeliveryStateEntry.DELIVERED_ONE ? normalize(deliveredLogin) : null);
            setNullableString(ps, 3, normalizeBlank(routesHash));
            setNullableLong(ps, 4, nextAttemptAtMs);
            setNullableString(ps, 5, normalizeBlank(lastError));
            ps.setLong(6, nowMs);
            ps.setString(7, current.getOutgoingMessageKey());
            ps.setString(8, current.getEventId());
            ps.executeUpdate();
        }
    }

    private MergeResult merge(
            int localState,
            String localLogin,
            String localHash,
            int remoteStateRaw,
            String remoteLoginRaw,
            String remoteHashRaw
    ) {
        int remoteState = normalizeState(remoteStateRaw, remoteLoginRaw);
        String remoteLogin = normalize(remoteLoginRaw);
        String localNormalizedLogin = normalize(localLogin);
        String localRoutesHash = normalizeBlank(localHash);
        String remoteRoutesHash = normalizeBlank(remoteHashRaw);
        String resultHash = remoteRoutesHash != null ? remoteRoutesHash : localRoutesHash;

        if (localState == DmDeliveryStateEntry.DELIVERED_ONE || localState == DmDeliveryStateEntry.DELIVERED_ALL) {
            return new MergeResult(DmDeliveryStateEntry.DELIVERED, localNormalizedLogin,
                    localRoutesHash != null ? localRoutesHash : remoteRoutesHash);
        }
        if (remoteState == DmDeliveryStateEntry.DELIVERED_ONE || remoteState == DmDeliveryStateEntry.DELIVERED_ALL) {
            return new MergeResult(DmDeliveryStateEntry.DELIVERED, remoteLogin, resultHash);
        }
        if (localState == DmDeliveryStateEntry.FAILED_FINAL || remoteState == DmDeliveryStateEntry.FAILED_FINAL) {
            return new MergeResult(DmDeliveryStateEntry.FAILED_FINAL, null, resultHash);
        }
        return new MergeResult(DmDeliveryStateEntry.PENDING_NONE, null, resultHash);
    }

    private int normalizeState(int state, String deliveredLogin) {
        if (state < DmDeliveryStateEntry.PENDING_NONE || state > DmDeliveryStateEntry.FAILED_FINAL) {
            return DmDeliveryStateEntry.PENDING_NONE;
        }
        if (state == DmDeliveryStateEntry.DELIVERED_ALL) return DmDeliveryStateEntry.DELIVERED;
        return state;
    }

    private String selectColumns() {
        return """
            SELECT outgoing_message_key, event_id, base_key, from_login, to_login,
                   incoming_message_key, created_at_ms, delivery_expires_at_ms,
                   delivery_state, delivered_server_login, recipient_routes_hash,
                   attempt_index, next_attempt_at_ms, last_attempt_at_ms, last_error, updated_at_ms
            FROM dm_delivery_state
            """;
    }

    private DmDeliveryStateEntry mapRow(ResultSet rs) throws SQLException {
        DmDeliveryStateEntry e = new DmDeliveryStateEntry();
        e.setOutgoingMessageKey(rs.getString("outgoing_message_key"));
        e.setEventId(rs.getString("event_id"));
        e.setBaseKey(rs.getString("base_key"));
        e.setFromLogin(rs.getString("from_login"));
        e.setToLogin(rs.getString("to_login"));
        e.setIncomingMessageKey(rs.getString("incoming_message_key"));
        e.setCreatedAtMs(rs.getLong("created_at_ms"));
        e.setDeliveryExpiresAtMs(rs.getLong("delivery_expires_at_ms"));
        e.setDeliveryState(rs.getInt("delivery_state"));
        e.setDeliveredServerLogin(rs.getString("delivered_server_login"));
        e.setRecipientRoutesHash(rs.getString("recipient_routes_hash"));
        e.setAttemptIndex(rs.getInt("attempt_index"));
        long next = rs.getLong("next_attempt_at_ms");
        e.setNextAttemptAtMs(rs.wasNull() ? null : next);
        long last = rs.getLong("last_attempt_at_ms");
        e.setLastAttemptAtMs(rs.wasNull() ? null : last);
        e.setLastError(rs.getString("last_error"));
        e.setUpdatedAtMs(rs.getLong("updated_at_ms"));
        return e;
    }

    private static void setNullableString(PreparedStatement ps, int index, String value) throws SQLException {
        if (isBlank(value)) ps.setNull(index, Types.VARCHAR);
        else ps.setString(index, value.trim());
    }

    private static void setNullableLong(PreparedStatement ps, int index, Long value) throws SQLException {
        if (value == null) ps.setNull(index, Types.BIGINT);
        else ps.setLong(index, value);
    }

    private static String normalize(String value) {
        if (value == null) return null;
        String normalized = value.trim().toLowerCase();
        return normalized.isEmpty() ? null : normalized;
    }

    private static String normalizeBlank(String value) {
        if (value == null) return null;
        String normalized = value.trim();
        return normalized.isEmpty() ? null : normalized;
    }

    private static boolean isBlank(String value) {
        return value == null || value.isBlank();
    }

    private record MergeResult(int state, String deliveredLogin, String routesHash) {}
}