import json
import time
import uuid
from typing import Any, Dict, List, Optional

import requests
from PyQt6.QtCore import QThread, pyqtSignal

from core.config import logger
from core.db import (
    SYNC_ACTION_ADD,
    SYNC_ACTION_DELETE,
    SYNC_ACTION_ROLLBACK,
    SYNC_ACTION_UPDATE,
    SyncJSONEncoder,
    sync_replay_mode,
)
from core.profile_sync_service import ProfileSyncService
from core.runtime_platform import detect_runtime_platform
from core.tenant_manager import TenantManager


def _is_safe_identifier(name: str) -> bool:
    return bool(name) and name.replace("_", "").isalnum()


class SyncWorker(QThread):
    PUSH_ENDPOINT = "https://app.golder.biz/api/sync/push/index.php"
    PULL_ENDPOINT = "https://app.golder.biz/api/sync/pull/index.php"
    events_applied = pyqtSignal(object)
    profile_deleted = pyqtSignal(object)
    sync_expired = pyqtSignal(str)

    def __init__(
        self,
        db,
        profile_id: str,
        parent=None,
        poll_interval_seconds: int = 10,
        request_timeout_seconds: int = 8,
    ):
        super().__init__(parent)
        self.db = db
        self.profile_id = str(profile_id or "")
        self.poll_interval_seconds = max(1, int(poll_interval_seconds or 10))
        self.request_timeout_seconds = max(2, int(request_timeout_seconds or 8))
        self._profile_sync_service = ProfileSyncService(
            db,
            tenant_manager=TenantManager(),
            request_timeout_seconds=max(8, self.request_timeout_seconds),
        )
        self._reconcile_interval_seconds = 60
        self._last_reconcile_monotonic = 0.0
        self._sync_disabled_reason = ""

    def run(self) -> None:
        while not self.isInterruptionRequested() and not self._sync_disabled_reason:
            if not self._ensure_sync_is_active():
                return
            try:
                self._run_profile_reconciliation_if_due()
            except Exception as e:
                logger.error(f"SyncWorker profile reconciliation failed: {e}", exc_info=True)

            try:
                self.push_to_vps()
            except Exception as e:
                logger.error(f"SyncWorker push failed: {e}", exc_info=True)

            try:
                self.pull_from_vps()
            except Exception as e:
                logger.error(f"SyncWorker pull failed: {e}", exc_info=True)

            for _ in range(self.poll_interval_seconds * 10):
                if self.isInterruptionRequested():
                    return
                self.msleep(100)

    def _ensure_sync_is_active(self) -> bool:
        tm = TenantManager()
        if tm.is_profile_sync_enabled():
            return True
        if tm.has_sync_entitlement() and tm.get_sync_status() == "expired":
            self._disable_sync("SYNC_EXPIRED")
            return False
        return False

    def _disable_sync(self, reason: str) -> None:
        normalized = str(reason or "SYNC_DISABLED").strip() or "SYNC_DISABLED"
        if self._sync_disabled_reason:
            return
        self._sync_disabled_reason = normalized
        logger.warning("Sync worker durduruldu: %s", normalized)
        if normalized == "SYNC_EXPIRED":
            try:
                self.sync_expired.emit(normalized)
            except Exception:
                pass
        self.requestInterruption()

    def _run_profile_reconciliation_if_due(self, force: bool = False) -> None:
        now = time.monotonic()
        if not force and (now - self._last_reconcile_monotonic) < self._reconcile_interval_seconds:
            return
        self._last_reconcile_monotonic = now
        self._profile_sync_service.run_profile_reconciliation()

    def _sync_profile_lifecycle_row(
        self,
        row: Dict[str, Any],
        payload: Dict[str, Any],
        profile_id: str,
        queue_id: int,
    ) -> bool:
        action_type = str(row.get("action_type") or "").upper().strip()
        tx_uuid = str(row.get("transaction_uuid") or uuid.uuid4())
        profile_name = str(
            ((payload.get("after") or {}) if isinstance(payload.get("after"), dict) else {}).get("name")
            or ((payload.get("before") or {}) if isinstance(payload.get("before"), dict) else {}).get("name")
            or ((payload.get("meta") or {}) if isinstance(payload.get("meta"), dict) else {}).get("profile_name")
            or ""
        ).strip()
        try:
            if action_type in {SYNC_ACTION_ADD, SYNC_ACTION_UPDATE}:
                self._profile_sync_service.push_profile_to_vps(
                    {"id": profile_id, "name": profile_name},
                    transaction_uuid=tx_uuid,
                )
                return True
            if action_type == SYNC_ACTION_DELETE:
                self._profile_sync_service.delete_remote_profile(
                    profile_id=profile_id,
                    profile_name=profile_name,
                    transaction_uuid=tx_uuid,
                )
                return True
        except Exception as exc:
            logger.warning(
                "Profile lifecycle queue row deferred (queue_id=%s, profile_id=%s): %s",
                queue_id,
                profile_id,
                exc,
            )
            return False
        return False

    def push_to_vps(self) -> None:
        if not self._ensure_sync_is_active():
            return
        state = self._get_or_create_sync_state()
        device_id = str(state.get("device_id") or "")
        tm = TenantManager()
        auth_headers = self._auth_headers(tm)
        if not auth_headers:
            logger.warning("Sync push skipped: API token is missing or expired.")
            return
        db_name = str(tm.load_license().get("db_name", "") or "")

        queue_rows = self.db.fetch_all(
            """
            SELECT id, transaction_uuid, table_name, action_type, record_id, payload_json
            FROM local_sync_queue
            ORDER BY id ASC
            LIMIT 200
            """
        ) or []
        if not queue_rows:
            return

        for row in queue_rows:
            if self.isInterruptionRequested():
                return

            queue_id = int(row.get("id") or 0)
            if queue_id <= 0:
                continue

            payload = self._parse_payload(row.get("payload_json"))
            table_name = str(row.get("table_name") or "")
            action_type = str(row.get("action_type") or "").upper().strip()
            profile_id = str(payload.get("profile_id") or self.profile_id)

            if table_name == "profiles":
                # Yeni sistemde profil degisiklikleri snapshot ile yonetiliyor.
                self._delete_queue_row(queue_id)
                continue

            req_body = {
                "queue_id": queue_id,
                "transaction_uuid": row.get("transaction_uuid"),
                "device_id": device_id,
                "profile_id": profile_id,
                "table_name": table_name,
                "action_type": row.get("action_type"),
                "record_id": row.get("record_id"),
                "payload_json": payload,
                "db_name": db_name,
            }
            res = self._post_json(self.PUSH_ENDPOINT, req_body, headers=auth_headers)
            status = str(res.get("status") or "").lower()
            code = str(res.get("code") or "")

            if status == "ok":
                self._delete_queue_row(queue_id)
                continue

            if code == "SYNC_EXPIRED":
                self._disable_sync(code)
                return

            # --- OTOMATİK KURTARMA VE TIKANIKLIK AÇMA KODU BAŞLANGICI ---
            if status == "deferred" and code == "PROFILE_NOT_READY":
                logger.warning(f"VPS'te profil ({profile_id}) hazır değil. Eski işlem atlanıyor.")
                self._delete_queue_row(queue_id)
                continue

            if status == "conflict" and code == "CONFLICT_REJECTED":
                rollback_error = None
                try:
                    self.rollback_local_transaction(
                        table_name=str(row.get("table_name") or ""),
                        action_type=str(row.get("action_type") or ""),
                        record_id=str(row.get("record_id") or ""),
                        payload=payload,
                        transaction_uuid=str(row.get("transaction_uuid") or ""),
                        profile_id=str(payload.get("profile_id") or self.profile_id),
                    )
                except Exception as e:
                    rollback_error = e
                conflict_message = "DİKKAT: Çakışma tespit edildi. İşlem iptal edildi!"
                if rollback_error is not None:
                    conflict_message = f"{conflict_message}\nYerel geri alma hatası: {rollback_error}"
                self._insert_conflict_log(conflict_message)
                self._delete_queue_row(queue_id)
                continue

            if status == "error":
                code = str(res.get("code") or "")
                
                # AĞ HATASI veya SUNUCU 5xx HATASI ise KESİNLİKLE SİLME! Döngüyü kır ve internet gelene kadar bekle.
                if code == "NETWORK_ERROR" or code.startswith("HTTP_5"):
                    logger.warning(f"Sync push ağ/sunucu hatası nedeniyle ertelendi (queue_id={queue_id}): {res}")
                    break
                
                # EĞER SUNUCU KESİN OLARAK İŞLEMİ REDDETTİYSE (Örn: 400 Format Hatası)
                # Sadece o zaman tıkanıklığı önlemek için sil.
                logger.error(f"Sync push sunucu tarafından kalıcı olarak reddedildi (queue_id={queue_id}): {res}. Tıkanıklığı önlemek için siliniyor.")
                self._delete_queue_row(queue_id)
                continue
            # --- OTOMATİK KURTARMA KODU BİTİŞİ ---

            logger.warning(f"Sync push ertelendi (Bağlantı kopuk olabilir) (queue_id={queue_id}): {res}")
            break
    def pull_from_vps(self) -> None:
        if not self._ensure_sync_is_active():
            return
        state = self._get_or_create_sync_state()
        last_pulled_event_id = self._get_profile_last_pulled_event_id(self.profile_id)
        device_id = str(state.get("device_id") or "")
        tm = TenantManager()
        auth_headers = self._auth_headers(tm)
        if not auth_headers:
            logger.warning("Sync pull skipped: API token is missing or expired.")
            return
        db_name = str(tm.load_license().get("db_name", "") or "")
        applied_events: List[Dict[str, Any]] = []
        active_profile_deleted = False

        req_body = {
            "profile_id": self.profile_id,
            "device_id": device_id,
            "last_pulled_event_id": last_pulled_event_id,
            "db_name": db_name,
        }
        res = self._post_json(self.PULL_ENDPOINT, req_body, headers=auth_headers)
        if str(res.get("code") or "") == "SYNC_EXPIRED":
            self._disable_sync("SYNC_EXPIRED")
            return
        if str(res.get("status") or "").lower() != "ok":
            logger.warning(f"Sync pull skipped: {res}")
            return

        events = res.get("events") or []
        if not isinstance(events, list):
            events = []

        events_sorted = sorted(events, key=lambda x: int((x or {}).get("id") or 0))

        for event in events_sorted:
            if self.isInterruptionRequested():
                return

            event_id = int((event or {}).get("id") or 0)
            if event_id <= last_pulled_event_id:
                continue

            table_name = str((event or {}).get("table_name") or "")
            action_type = str((event or {}).get("action_type") or "").upper().strip()
            record_id = str((event or {}).get("record_id") or "")
            tx_uuid = str((event or {}).get("transaction_uuid") or "")
            payload = self._parse_payload((event or {}).get("payload_json"))
            if table_name == "profiles":
                event_profile_id = str(
                    payload.get("profile_id")
                    or (event or {}).get("profile_id")
                    or record_id
                    or self.profile_id
                )
            else:
                event_profile_id = str(payload.get("profile_id") or (event or {}).get("profile_id") or self.profile_id)

            if table_name == "profiles" and action_type != SYNC_ACTION_ROLLBACK:
                self._handle_remote_profile_event(
                    action_type=action_type,
                    record_id=record_id,
                    payload=payload,
                    event_profile_id=event_profile_id,
                    transaction_uuid=tx_uuid,
                )
                if action_type == SYNC_ACTION_DELETE and event_profile_id == self.profile_id:
                    active_profile_deleted = True
            else:
                with self.db.transaction_scope() as cursor:
                    with sync_replay_mode(True):
                        if action_type == SYNC_ACTION_ROLLBACK:
                            self.rollback_local_transaction(
                                table_name=table_name,
                                action_type=str((payload.get("meta") or {}).get("original_action") or ""),
                                record_id=record_id,
                                payload=payload,
                                transaction_uuid=tx_uuid,
                                profile_id=event_profile_id,
                                cursor=cursor,
                            )
                        else:
                            self.apply_remote_event(
                                table_name=table_name,
                                action_type=action_type,
                                record_id=record_id,
                                payload=payload,
                                profile_id=event_profile_id,
                                cursor=cursor,
                            )

                    cursor.execute(
                        """
                        INSERT INTO sync_profile_state (profile_id, last_pulled_event_id)
                        VALUES (%s, %s)
                        ON DUPLICATE KEY UPDATE last_pulled_event_id = VALUES(last_pulled_event_id)
                        """,
                        (self.profile_id, event_id),
                    )

            if table_name == "profiles" and action_type != SYNC_ACTION_ROLLBACK:
                self._set_profile_last_pulled_event_id(self.profile_id, event_id)

            last_pulled_event_id = event_id
            applied_events.append(
                {
                    "id": event_id,
                    "profile_id": event_profile_id,
                    "table_name": table_name,
                    "action_type": action_type,
                    "record_id": record_id,
                    "payload_json": payload,
                }
            )
            if active_profile_deleted:
                break

        if bool(res.get("profile_missing")) and self._profile_sync_service.has_local_profile(self.profile_id):
            try:
                if not self._profile_sync_service.has_local_tombstone(self.profile_id):
                    self._profile_sync_service.push_profile_to_vps(
                        {"id": self.profile_id},
                        transaction_uuid=str(uuid.uuid4()),
                    )
            except Exception as exc:
                logger.warning(
                    "Remote profile missing snapshot push deferred (profile_id=%s): %s",
                    self.profile_id,
                    exc,
                )

        if applied_events:
            try:
                self.events_applied.emit(applied_events)
            except Exception:
                pass

    def _handle_remote_profile_event(
        self,
        action_type: str,
        record_id: str,
        payload: Dict[str, Any],
        event_profile_id: str,
        transaction_uuid: str,
    ) -> None:
        profile_id = str(event_profile_id or record_id or "").strip()
        if not profile_id:
            return
        act = str(action_type or "").upper().strip()
        before_row = payload.get("before") if isinstance(payload, dict) else None
        after_row = payload.get("after") if isinstance(payload, dict) else None
        profile_name = str(
            ((after_row or {}) if isinstance(after_row, dict) else {}).get("name")
            or ((before_row or {}) if isinstance(before_row, dict) else {}).get("name")
            or ""
        ).strip()

        if act in {SYNC_ACTION_ADD, SYNC_ACTION_UPDATE}:
            if self._profile_sync_service.has_local_tombstone(profile_id):
                return
            if (not self._profile_sync_service.has_local_profile(profile_id)) or (
                not self._profile_sync_service.is_local_profile_initialized(profile_id)
            ):
                self._profile_sync_service.pull_profile_to_local(profile_id, transaction_uuid=transaction_uuid)
                return
            with self.db.transaction_scope() as cursor:
                with sync_replay_mode(True):
                    self.apply_remote_event(
                        table_name="profiles",
                        action_type=act,
                        record_id=profile_id,
                        payload=payload,
                        profile_id=profile_id,
                        cursor=cursor,
                    )
            return

        if act == SYNC_ACTION_DELETE:
            deleted = self._profile_sync_service.apply_remote_profile_deletion(
                profile_id=profile_id,
                profile_name=profile_name,
                transaction_uuid=transaction_uuid,
            )
            self._emit_profile_deleted(deleted)

    def _emit_profile_deleted(self, payload: Optional[Dict[str, Any]]) -> None:
        if not isinstance(payload, dict):
            return
        try:
            self.profile_deleted.emit(payload)
        except Exception:
            pass

    def apply_remote_event(
        self,
        table_name: str,
        action_type: str,
        record_id: str,
        payload: Dict[str, Any],
        profile_id: str,
        cursor,
    ) -> None:
        if not _is_safe_identifier(table_name):
            return

        act = str(action_type or "").upper().strip()
        before_row = payload.get("before") if isinstance(payload, dict) else None
        after_row = payload.get("after") if isinstance(payload, dict) else None

        if act in {SYNC_ACTION_ADD, SYNC_ACTION_UPDATE}:
            if isinstance(after_row, dict) and after_row:
                self._upsert_row(table_name, after_row, profile_id, cursor, fallback_id=record_id)
            return

        if act == SYNC_ACTION_DELETE:
            if isinstance(after_row, dict) and after_row:
                self._upsert_row(table_name, after_row, profile_id, cursor, fallback_id=record_id)
            else:
                target_id = record_id or str((before_row or {}).get("id") or "")
                row_hint = before_row if isinstance(before_row, dict) and before_row else None
                self._soft_delete_or_delete_row(table_name, target_id, profile_id, cursor, row_hint=row_hint)

    def rollback_local_transaction(
        self,
        table_name: str,
        action_type: str,
        record_id: str,
        payload: Dict[str, Any],
        transaction_uuid: str = "",
        profile_id: str = "",
        cursor=None,
    ) -> None:
        if not _is_safe_identifier(table_name):
            return

        effective_profile = str(profile_id or payload.get("profile_id") or self.profile_id)
        act = str(action_type or "").upper().strip()
        before_row = payload.get("before") if isinstance(payload, dict) else None
        after_row = payload.get("after") if isinstance(payload, dict) else None

        def _apply(cur) -> None:
            with sync_replay_mode(True):
                if act == SYNC_ACTION_ADD:
                    target_id = record_id or str((after_row or {}).get("id") or "")
                    row_hint = after_row if isinstance(after_row, dict) and after_row else None
                    self._soft_delete_or_delete_row(
                        table_name,
                        target_id,
                        effective_profile,
                        cur,
                        row_hint=row_hint,
                    )
                    return
                if act in {SYNC_ACTION_UPDATE, SYNC_ACTION_DELETE}:
                    if isinstance(before_row, dict) and before_row:
                        self._upsert_row(
                            table_name,
                            before_row,
                            effective_profile,
                            cur,
                            fallback_id=record_id,
                        )
                    return
                if act == SYNC_ACTION_ROLLBACK and isinstance(before_row, dict) and before_row:
                    self._upsert_row(
                        table_name,
                        before_row,
                        effective_profile,
                        cur,
                        fallback_id=record_id,
                    )

        try:
            if cursor is not None:
                _apply(cursor)
            else:
                with self.db.transaction_scope() as cur:
                    _apply(cur)
        except Exception as e:
            logger.error(
                f"Rollback failed (tx_uuid={transaction_uuid}, table={table_name}, record={record_id}): {e}",
                exc_info=True,
            )
            raise

    def _upsert_row(
        self,
        table_name: str,
        row: Dict[str, Any],
        profile_id: str,
        cursor,
        fallback_id: str = "",
    ) -> None:
        if not _is_safe_identifier(table_name):
            return
        if not isinstance(row, dict) or not row:
            return

        columns = self.db.get_table_columns(table_name) or set()
        if not columns:
            logger.warning(f"Sync upsert skipped: table metadata unavailable ({table_name})")
            return
        safe_data: Dict[str, Any] = {}
        has_id_col = "id" in columns
        row_id = str(row.get("id") or fallback_id or "")
        if has_id_col and not row_id:
            return

        for key, value in row.items():
            if not _is_safe_identifier(str(key)):
                continue
            if key not in columns:
                continue
            safe_data[str(key)] = value

        if has_id_col:
            safe_data["id"] = row_id
        if "profile_id" in columns and not safe_data.get("profile_id"):
            safe_data["profile_id"] = profile_id
        if not safe_data:
            return

        col_names = list(safe_data.keys())
        placeholders = ", ".join(["%s"] * len(col_names))
        columns_sql = ", ".join([f"`{c}`" for c in col_names])
        update_cols = [c for c in col_names if c != "id"]
        sql = f"INSERT INTO `{table_name}` ({columns_sql}) VALUES ({placeholders})"
        if update_cols:
            update_sql = ", ".join([f"`{c}` = VALUES(`{c}`)" for c in update_cols])
            sql += f" ON DUPLICATE KEY UPDATE {update_sql}"

        cursor.execute(sql, tuple(safe_data[c] for c in col_names))

    def _soft_delete_or_delete_row(
        self,
        table_name: str,
        record_id: str,
        profile_id: str,
        cursor,
        row_hint: Optional[Dict[str, Any]] = None,
    ) -> None:
        if not _is_safe_identifier(table_name):
            return

        has_id = self.db.has_column(table_name, "id")
        has_profile = self.db.has_column(table_name, "profile_id")
        has_soft_delete = self.db.has_column(table_name, "is_deleted")
        target_id = str(record_id or "").strip()

        if has_id and target_id:
            if has_soft_delete:
                if has_profile:
                    cursor.execute(
                        f"UPDATE `{table_name}` SET is_deleted = 1 WHERE id = %s AND profile_id = %s",
                        (target_id, profile_id),
                    )
                else:
                    cursor.execute(
                        f"UPDATE `{table_name}` SET is_deleted = 1 WHERE id = %s",
                        (target_id,),
                    )
                return
            if has_profile:
                cursor.execute(
                    f"DELETE FROM `{table_name}` WHERE id = %s AND profile_id = %s",
                    (target_id, profile_id),
                )
            else:
                cursor.execute(f"DELETE FROM `{table_name}` WHERE id = %s", (target_id,))
            return

        # Fallback for id-less sync tables.
        if not isinstance(row_hint, dict):
            return
        if table_name == "user_permissions":
            user_id = str(row_hint.get("user_id") or "").strip()
            perm_code = str(row_hint.get("permission_code") or "").strip()
            if not user_id or not perm_code:
                return
            cursor.execute(
                "DELETE FROM `user_permissions` WHERE user_id = %s AND permission_code = %s",
                (user_id, perm_code),
            )
            return

        if table_name == "user_access_scopes":
            user_id = str(row_hint.get("user_id") or "").strip()
            scope_type = str(row_hint.get("scope_type") or "").strip()
            target = str(row_hint.get("target_id") or "").strip()
            if not user_id or not scope_type or not target:
                return
            cursor.execute(
                "DELETE FROM `user_access_scopes` WHERE user_id = %s AND scope_type = %s AND target_id = %s",
                (user_id, scope_type, target),
            )

    def _post_json(
        self,
        url: str,
        payload: Dict[str, Any],
        headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        try:
            is_mac = detect_runtime_platform() == "macos"
            verify_ssl = not is_mac
            if is_mac:
                import urllib3

                urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
            req_headers = dict(headers or {})
            if is_mac:
                req_headers["User-Agent"] = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
            json_payload = self._json_ready_payload(payload)
            safe_timeout = max(20 if is_mac else 10, self.request_timeout_seconds)
            response = requests.post(
                url,
                json=json_payload,
                headers=req_headers,
                timeout=safe_timeout,
                allow_redirects=False,
                verify=verify_ssl,
            )
            if response.status_code in (301, 302, 307, 308):
                location = str(response.headers.get("Location") or "").strip()
                if location:
                    redirect_url = requests.compat.urljoin(url, location)
                    response = requests.post(
                        redirect_url,
                        json=json_payload,
                        headers=req_headers,
                        timeout=safe_timeout,
                        allow_redirects=False,
                        verify=verify_ssl,
                    )
            if response.status_code == 405 and not str(url).endswith("/"):
                retry_url = f"{url}/"
                response = requests.post(
                    retry_url,
                    json=json_payload,
                    headers=req_headers,
                    timeout=safe_timeout,
                    allow_redirects=False,
                    verify=verify_ssl,
                )
            try:
                data = response.json()
            except Exception:
                data = {"status": "error", "code": "INVALID_JSON", "text": response.text}
            if isinstance(data, dict):
                data.setdefault("http_status", response.status_code)
            if response.status_code >= 500:
                return {
                    "status": "error",
                    "code": f"HTTP_{response.status_code}",
                    "message": str(data),
                }
            return data if isinstance(data, dict) else {"status": "error", "code": "INVALID_PAYLOAD"}
        except Exception as e:
            return {"status": "error", "code": "NETWORK_ERROR", "message": str(e)}

    def _json_ready_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        if not isinstance(payload, dict):
            return {}
        try:
            encoded = json.dumps(payload, cls=SyncJSONEncoder, ensure_ascii=False)
            decoded = json.loads(encoded)
            return decoded if isinstance(decoded, dict) else {}
        except Exception:
            logger.warning("SyncWorker payload normalization failed; falling back to stringified values.")
            return {
                str(key): str(value)
                for key, value in payload.items()
                if isinstance(key, str)
            }

    def _parse_payload(self, payload_json: Any) -> Dict[str, Any]:
        if isinstance(payload_json, dict):
            return dict(payload_json)
        if isinstance(payload_json, str):
            try:
                parsed = json.loads(payload_json)
                return parsed if isinstance(parsed, dict) else {}
            except Exception:
                return {}
        return {}

    def _delete_queue_row(self, queue_id: int) -> None:
        try:
            self.db.execute_non_query("DELETE FROM local_sync_queue WHERE id = %s", (queue_id,))
        except Exception as e:
            logger.error(f"Could not delete local_sync_queue row ({queue_id}): {e}", exc_info=True)

    def _insert_conflict_log(self, message: str) -> None:
        try:
            self.db.execute_non_query(
                "INSERT INTO conflict_logs (message, is_read, created_at) VALUES (%s, 0, NOW())",
                (message,),
            )
        except Exception as e:
            logger.error(f"Could not insert conflict log: {e}", exc_info=True)

    def _get_or_create_sync_state(self) -> Dict[str, Any]:
        row = self.db.fetch_one("SELECT last_pulled_event_id, device_id FROM sync_state LIMIT 1")
        if row and row.get("device_id"):
            return row

        new_device_id = str(uuid.uuid4())
        try:
            with self.db.transaction_scope() as cursor:
                cursor.execute(
                    "SELECT last_pulled_event_id, device_id FROM sync_state LIMIT 1 FOR UPDATE"
                )
                locked = cursor.fetchone()
                if locked and locked.get("device_id"):
                    return locked
                cursor.execute(
                    "INSERT INTO sync_state (last_pulled_event_id, device_id) VALUES (%s, %s)",
                    (0, new_device_id),
                )
        except Exception:
            pass
        return {"last_pulled_event_id": 0, "device_id": new_device_id}

    def _get_profile_last_pulled_event_id(self, profile_id: str) -> int:
        pid = str(profile_id or "").strip()
        if not pid:
            return 0
        try:
            row = self.db.fetch_one(
                "SELECT last_pulled_event_id FROM sync_profile_state WHERE profile_id = %s LIMIT 1",
                (pid,),
            )
            return int((row or {}).get("last_pulled_event_id") or 0)
        except Exception:
            return 0

    def _set_profile_last_pulled_event_id(self, profile_id: str, event_id: int) -> None:
        pid = str(profile_id or "").strip()
        if not pid or int(event_id or 0) <= 0:
            return
        try:
            self.db.execute_non_query(
                """
                INSERT INTO sync_profile_state (profile_id, last_pulled_event_id)
                VALUES (%s, %s)
                ON DUPLICATE KEY UPDATE last_pulled_event_id = VALUES(last_pulled_event_id)
                """,
                (pid, int(event_id)),
            )
        except Exception as e:
            logger.error(f"Could not update sync_profile_state ({pid}, {event_id}): {e}", exc_info=True)

    def _auth_headers(self, tenant_manager: Optional[TenantManager] = None) -> Dict[str, str]:
        tm = tenant_manager or TenantManager()
        token = tm.get_api_token()
        if not token:
            return {}
        return {"Authorization": f"Bearer {token}"}




