from __future__ import annotations

import json
import uuid
from datetime import datetime
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union

try:
    import requests
except Exception:  # pragma: no cover - optional dependency on runtime
    requests = None  # type: ignore[assignment]
import urllib3

from core.config import logger
from core.db import SYNC_ACTION_ADD, SYNC_ACTION_DELETE, SyncJSONEncoder, queue_row_sync_event
from core.runtime_platform import detect_runtime_platform
from core.tenant_manager import TenantManager


PROFILE_SYNC_EXCLUDED_TABLES = {
    "conflict_logs",
    "deleted_profiles",
    "global_sync_events",
    "local_sync_queue",
    "profile_sync_jobs",
    "sync_profile_state",
    "sync_state",
}

PROFILE_SYNC_PRIORITY_TABLES = [
    "app_settings",
    "document_counters",
    "users",
    "depots",
    "categories",
    "persons",
    "product_definitions",
    "financial_accounts",
    "taxes",
    "tax_profiles",
    "tax_profile_items",
    "tax_applications",
    "stock_items",
    "barcode_history",
    "stock_movements",
    "financial_transactions",
    "sales",
    "debt_records",
    "customer_history_events",
    "repair_records",
    "scheduled_transactions",
    "service_definitions",
    "service_jobs",
    "agenda_notes",
    "audit_logs",
]


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


def _utc_sql_now() -> str:
    return datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")


def _normalize_profile_name(name: str) -> str:
    return " ".join(str(name or "").strip().split()).casefold()


def _normalize_row(row: Dict[str, Any]) -> Dict[str, Any]:
    normalized: Dict[str, Any] = {}
    for key, value in (row or {}).items():
        if isinstance(key, str) and _is_safe_identifier(key):
            normalized[key] = value
    return normalized


def _order_profile_tables(table_names: Iterable[str]) -> List[str]:
    remaining = {name for name in table_names if name}
    ordered: List[str] = []
    for name in PROFILE_SYNC_PRIORITY_TABLES:
        if name in remaining:
            ordered.append(name)
            remaining.remove(name)
    ordered.extend(sorted(remaining))
    return ordered


class ProfileSyncService:
    CATALOG_ENDPOINT = "https://app.golder.biz/api/sync/profiles/catalog.php"
    EXPORT_ENDPOINT = "https://app.golder.biz/api/sync/profiles/export.php"
    IMPORT_ENDPOINT = "https://app.golder.biz/api/sync/profiles/import.php"
    DELETE_ENDPOINT = "https://app.golder.biz/api/sync/profiles/delete.php"
    PUSH_ENDPOINT = "https://app.golder.biz/api/sync/push/index.php"

    def __init__(
        self,
        db,
        tenant_manager: Optional[TenantManager] = None,
        request_timeout_seconds: int = 20,
    ) -> None:
        self.db = db
        self.tenant_manager = tenant_manager or TenantManager()
        self.request_timeout_seconds = max(5, int(request_timeout_seconds or 20))

    # Public API
    def fetch_local_profiles(self) -> Dict[str, Dict[str, Any]]:
        updated_at_expr = (
            "COALESCE(updated_at, created_at, NOW())"
            if self.db.has_column("profiles", "updated_at")
            else "COALESCE(created_at, NOW())"
        )
        rows = self.db.fetch_all(
            f"""
            SELECT id, name, is_active, COALESCE(is_deleted, 0) AS is_deleted,
                   {updated_at_expr} AS updated_at
            FROM profiles
            WHERE COALESCE(is_deleted, 0) = 0
            ORDER BY name ASC
            """
        ) or []
        return self._profile_rows_to_map(rows)

    def fetch_vps_profiles(self) -> Dict[str, Dict[str, Any]]:
        remote_profiles, _ = self._fetch_remote_catalog()
        return remote_profiles

    def detect_missing_local(
        self,
        local_profiles: Dict[str, Dict[str, Any]],
        vps_profiles: Dict[str, Dict[str, Any]],
    ) -> List[Dict[str, Any]]:
        return [dict(vps_profiles[pid]) for pid in sorted(vps_profiles.keys()) if pid not in local_profiles]

    def detect_missing_remote(
        self,
        local_profiles: Dict[str, Dict[str, Any]],
        vps_profiles: Dict[str, Dict[str, Any]],
    ) -> List[Dict[str, Any]]:
        return [dict(local_profiles[pid]) for pid in sorted(local_profiles.keys()) if pid not in vps_profiles]

    def push_profile_to_vps(
        self,
        profile: Dict[str, Any],
        transaction_uuid: Optional[str] = None,
    ) -> Dict[str, Any]:
        profile_id = str((profile or {}).get("id") or "").strip()
        if not profile_id:
            raise RuntimeError("Remote profile push requires profile id.")
        if self._has_local_tombstone(profile_id):
            raise RuntimeError(f"Profile {profile_id} is tombstoned locally and cannot be re-uploaded.")
        snapshot = self._export_local_profile(profile_id)
        payload = self._import_remote_profile(
            snapshot=snapshot,
            transaction_uuid=transaction_uuid or str(uuid.uuid4()),
        )
        self._set_local_profile_cursor(profile_id, int(payload.get("max_event_id") or 0))
        return payload

    def pull_profile_to_local(
        self,
        profile: Union[str, Dict[str, Any]],
        transaction_uuid: Optional[str] = None,
    ) -> Dict[str, Any]:
        profile_id = str(profile if isinstance(profile, str) else (profile or {}).get("id") or "").strip()
        if not profile_id:
            raise RuntimeError("Local profile pull requires profile id.")
        if self._has_local_tombstone(profile_id):
            raise RuntimeError(f"Profile {profile_id} is tombstoned locally and cannot be auto-restored.")
        export_payload = self._export_remote_profile(profile_id)
        snapshot = export_payload.get("snapshot") or {}
        if transaction_uuid:
            snapshot.setdefault("_transaction_uuid", transaction_uuid)
        self._import_local_snapshot(snapshot)
        self._set_local_profile_cursor(profile_id, int(export_payload.get("max_event_id") or 0))
        return export_payload

    def run_profile_reconciliation(self) -> bool:
        """
        YENİ, SADE VE KUSURSUZ SENKRONİZASYON MANTIĞI:
        Kuyruk(Queue) kullanmaz. İki tarafın (Lokal ve VPS) kataloglarını karşılaştırır 
        ve eksik/silinmiş olanları anında eşitler.
        """
        if not self.tenant_manager.is_profile_sync_enabled():
            return False
        
        try:
            # 1. LOKAL DURUMU AL
            local_rows = self.db.fetch_all("SELECT id, name, COALESCE(is_deleted, 0) as is_deleted FROM profiles") or []
            local_active = {r['id']: r for r in local_rows if r['is_deleted'] == 0}
            local_deleted_rows = self.db.fetch_all("SELECT profile_id FROM deleted_profiles") or []
            local_deleted = [r['profile_id'] for r in local_deleted_rows]

            # 2. VPS DURUMUNU AL
            remote_data = self._request_json("GET", self.CATALOG_ENDPOINT)
            remote_active = {r['id']: r for r in remote_data.get('profiles', [])}
            remote_deleted = [r['profile_id'] for r in remote_data.get('deleted_profiles', [])]

            self._resolve_cross_catalog_name_conflicts(local_active, remote_active)
            local_active = self.fetch_local_profiles()
            remote_active = self.fetch_vps_profiles()

            # 3. KARŞILAŞTIR VE UYGULA

            # KURAL 1: VPS'te silinmişse, Lokalde de SİL
            for pid in remote_deleted:
                if pid in local_active:
                    logger.info(f"[SYNC] Profil ({local_active[pid]['name']}) VPS'te silinmiş. Lokalde de siliniyor...")
                    self.delete_local_profile(pid, queue_remote_delete=False)

            # KURAL 2: Lokalde silinmişse, VPS'te de SİL
            for pid in local_deleted:
                if pid in remote_active:
                    logger.info(f"[SYNC] Profil ({remote_active[pid]['name']}) lokalde silinmiş. VPS'ten siliniyor...")
                    self.delete_remote_profile(profile_id=pid)

            # Katalogları güncelle (Silinenleri aradan çıkardık)
            local_keys = set(local_active.keys()) - set(remote_deleted)
            remote_keys = set(remote_active.keys()) - set(local_deleted)

            # KURAL 3: Lokalde var, VPS'te yok -> VPS'e YÜKLE (Snapshot Push)
            for pid in local_keys:
                if pid not in remote_active:
                    logger.info(f"[SYNC] Profil ({local_active[pid]['name']}) VPS'te yok. VPS'e gönderiliyor...")
                    self.push_profile_to_vps(local_active[pid])

            # KURAL 4: VPS'te var, Lokalde yok -> Lokale İNDİR (Snapshot Pull)
            for pid in remote_keys:
                if pid not in local_active:
                    logger.info(f"[SYNC] Profil ({remote_active[pid]['name']}) lokalde yok. VPS'ten indiriliyor...")
                    self.pull_profile_to_local(pid)

            return True
        except Exception as exc:
            logger.error(f"[SYNC] Profil eşitleme başarısız oldu: {exc}")
            return False

    # Catalog helpers
    def _fetch_remote_catalog(self) -> Tuple[Dict[str, Dict[str, Any]], Dict[str, Dict[str, Any]]]:
        payload = self._request_json("GET", self.CATALOG_ENDPOINT)
        profiles = self._profile_rows_to_map(payload.get("profiles") or [])
        tombstones = self._tombstone_rows_to_map(payload.get("deleted_profiles") or [])
        return profiles, tombstones

    def _fetch_local_tombstones(self) -> Dict[str, Dict[str, Any]]:
        rows = self.db.fetch_all(
            """
            SELECT profile_id, profile_name, transaction_uuid, device_id, deleted_at
            FROM deleted_profiles
            ORDER BY deleted_at ASC
            """
        ) or []
        return self._tombstone_rows_to_map(rows)

    def _profile_rows_to_map(self, rows: Iterable[Any]) -> Dict[str, Dict[str, Any]]:
        result: Dict[str, Dict[str, Any]] = {}
        for row in rows or []:
            if not isinstance(row, dict):
                continue
            profile_id = str(row.get("id") or row.get("profile_id") or "").strip()
            if not profile_id:
                continue
            result[profile_id] = {
                "id": profile_id,
                "name": str(row.get("name") or "").strip(),
                "is_active": int(row.get("is_active") or 1),
                "is_deleted": int(row.get("is_deleted") or 0),
                "updated_at": str(row.get("updated_at") or ""),
            }
        return result

    def _tombstone_rows_to_map(self, rows: Iterable[Any]) -> Dict[str, Dict[str, Any]]:
        result: Dict[str, Dict[str, Any]] = {}
        for row in rows or []:
            if not isinstance(row, dict):
                continue
            profile_id = str(row.get("profile_id") or row.get("id") or "").strip()
            if not profile_id:
                continue
            result[profile_id] = {
                "profile_id": profile_id,
                "profile_name": str(row.get("profile_name") or row.get("name") or "").strip(),
                "transaction_uuid": str(row.get("transaction_uuid") or "").strip(),
                "device_id": str(row.get("device_id") or "").strip(),
                "deleted_at": str(row.get("deleted_at") or ""),
            }
        return result

    # Local lifecycle helpers
    def create_local_profile(self, profile_name: str, source: str = "profile.create") -> Dict[str, Any]:
        name = str(profile_name or "").strip()
        if not name: raise RuntimeError("Profile name is required.")
        
        exists = self.db.fetch_one("SELECT id FROM profiles WHERE name = %s AND COALESCE(is_deleted, 0) = 0 LIMIT 1", (name,))
        if exists: raise RuntimeError("Profile already exists.")

        profile_id = str(uuid.uuid4())
        admin_user_id = str(uuid.uuid4())
        
        with self.db.transaction_scope() as cursor:
            # Sadece profili oluştur, kuyruğa(queue) yazma. Yeni beyin bunu görecek ve VPS'e atacak!
            cursor.execute("INSERT INTO profiles (id, name, is_active, is_deleted) VALUES (%s, %s, 1, 0)", (profile_id, name))
            self._ensure_default_local_profile_settings(profile_id, cursor)
            cursor.execute(
                """
                INSERT INTO users (id, profile_id, username, full_name, password_hash, role, is_active, is_deleted)
                VALUES (%s, %s, 'admin', 'Yönetici', '', 'admin', 1, 0)
                """,
                (admin_user_id, profile_id),
            )
            
        return {"id": profile_id, "name": name}

    def delete_local_profile(self, profile_id: str, profile_name: str = "", source: str = "", queue_remote_delete: bool = True, transaction_uuid: Optional[str] = None) -> Dict[str, Any]:
        pid = str(profile_id or "").strip()
        
        with self.db.transaction_scope() as cursor:
            # Sadece silindi(tombstone) olarak işaretle ve verilerini temizle.
            self._record_local_tombstone(profile_id=pid, profile_name=profile_name, transaction_uuid=str(uuid.uuid4()), cursor=cursor)
            self._purge_local_profile_queue(pid, cursor)
            self._clear_local_profile_sync_state(pid, cursor)
            self._delete_profile_with_cursor(pid, cursor)

        return {"profile_id": pid}

    def has_local_profile(self, profile_id: str) -> bool:
        pid = str(profile_id or "").strip()
        if not pid:
            return False
        row = self.db.fetch_one(
            """
            SELECT id
            FROM profiles
            WHERE id = %s AND COALESCE(is_deleted, 0) = 0
            LIMIT 1
            """,
            (pid,),
        )
        return bool(row)

    def has_local_tombstone(self, profile_id: str) -> bool:
        return self._has_local_tombstone(profile_id)

    def is_local_profile_initialized(self, profile_id: str) -> bool:
        pid = str(profile_id or "").strip()
        if not pid or self._has_local_tombstone(pid):
            return False
        row = self.db.fetch_one(
            """
            SELECT profile_id
            FROM app_settings
            WHERE profile_id = %s
            LIMIT 1
            """,
            (pid,),
        )
        return bool(row)

    def apply_remote_profile_deletion(
        self,
        profile_id: str,
        profile_name: str = "",
        transaction_uuid: Optional[str] = None,
    ) -> Dict[str, Any]:
        pid = str(profile_id or "").strip()
        if not pid:
            raise RuntimeError("profile_id is required.")
        tx_uuid = str(transaction_uuid or uuid.uuid4())
        if not self.has_local_profile(pid):
            with self.db.transaction_scope() as cursor:
                self._record_local_tombstone(
                    profile_id=pid,
                    profile_name=str(profile_name or "").strip(),
                    transaction_uuid=tx_uuid,
                    cursor=cursor,
                )
                self._purge_local_profile_queue(pid, cursor)
                self._clear_local_profile_sync_state(pid, cursor)
            return {
                "profile_id": pid,
                "profile_name": str(profile_name or "").strip(),
                "transaction_uuid": tx_uuid,
                "already_deleted": True,
            }
        return self.delete_local_profile(
            profile_id=pid,
            profile_name=profile_name,
            source="sync.remote_profile_delete",
            queue_remote_delete=False,
            transaction_uuid=tx_uuid,
        )

    def delete_remote_profile(
        self,
        profile_id: str,
        profile_name: str = "",
        transaction_uuid: Optional[str] = None,
    ) -> Dict[str, Any]:
        pid = str(profile_id or "").strip()
        if not pid:
            raise RuntimeError("profile_id is required.")
        return self._request_json(
            "POST",
            self.DELETE_ENDPOINT,
            {
                "profile_id": pid,
                "profile_name": str(profile_name or "").strip(),
                "device_id": self._get_or_create_device_id(),
                "transaction_uuid": str(transaction_uuid or uuid.uuid4()),
            },
        )

    # Reconciliation helpers
    def _reconcile_deletions(
        self,
        local_profiles: Dict[str, Dict[str, Any]],
        remote_profiles: Dict[str, Dict[str, Any]],
        local_tombstones: Dict[str, Dict[str, Any]],
        remote_tombstones: Dict[str, Dict[str, Any]],
    ) -> bool:
        had_errors = False

        for profile_id, tombstone in sorted(remote_tombstones.items()):
            if profile_id not in local_profiles:
                continue
            try:
                self.apply_remote_profile_deletion(
                    profile_id=profile_id,
                    profile_name=str(tombstone.get("profile_name") or ""),
                    transaction_uuid=str(tombstone.get("transaction_uuid") or uuid.uuid4()),
                )
            except Exception as exc:
                logger.error("Remote tombstone apply failed for profile_id=%s: %s", profile_id, exc, exc_info=True)
                had_errors = True

        for profile_id, tombstone in sorted(local_tombstones.items()):
            if profile_id not in remote_profiles:
                continue
            try:
                self.delete_remote_profile(
                    profile_id=profile_id,
                    profile_name=str(tombstone.get("profile_name") or ""),
                    transaction_uuid=str(tombstone.get("transaction_uuid") or uuid.uuid4()),
                )
            except Exception as exc:
                logger.error("Remote tombstone push failed for profile_id=%s: %s", profile_id, exc, exc_info=True)
                had_errors = True

        return had_errors

    def _reconcile_shared_profiles(
        self,
        local_profiles: Dict[str, Dict[str, Any]],
        remote_profiles: Dict[str, Dict[str, Any]],
    ) -> bool:
        had_errors = False
        shared_ids = sorted(set(local_profiles.keys()) & set(remote_profiles.keys()))
        for profile_id in shared_ids:
            local_row = dict(local_profiles.get(profile_id) or {})
            remote_row = dict(remote_profiles.get(profile_id) or {})
            if not local_row or not remote_row:
                continue
            local_sig = (
                local_row.get("name"),
                int(local_row.get("is_active") or 1),
                int(local_row.get("is_deleted") or 0),
            )
            remote_sig = (
                remote_row.get("name"),
                int(remote_row.get("is_active") or 1),
                int(remote_row.get("is_deleted") or 0),
            )
            if local_sig == remote_sig:
                continue
            try:
                self._upsert_local_profile(remote_row)
            except Exception as exc:
                logger.error("Shared profile metadata reconcile failed for profile_id=%s: %s", profile_id, exc)
                had_errors = True
        return had_errors

    def _resolve_cross_catalog_name_conflicts(
        self,
        local_profiles: Dict[str, Dict[str, Any]],
        remote_profiles: Dict[str, Dict[str, Any]],
    ) -> bool:
        had_errors = False
        combined: Dict[str, Dict[str, Any]] = {}
        for profile_id, row in local_profiles.items():
            combined.setdefault(profile_id, {})["local"] = dict(row)
        for profile_id, row in remote_profiles.items():
            combined.setdefault(profile_id, {})["remote"] = dict(row)

        groups: Dict[str, List[str]] = {}
        for profile_id, sides in combined.items():
            name = ""
            if isinstance(sides.get("local"), dict):
                name = str((sides.get("local") or {}).get("name") or "")
            elif isinstance(sides.get("remote"), dict):
                name = str((sides.get("remote") or {}).get("name") or "")
            normalized = _normalize_profile_name(name)
            if normalized:
                groups.setdefault(normalized, []).append(profile_id)

        all_rows: Dict[str, Dict[str, Any]] = {}
        for profile_id, row in local_profiles.items():
            all_rows[profile_id] = dict(row)
        for profile_id, row in remote_profiles.items():
            all_rows.setdefault(profile_id, dict(row))

        for _, profile_ids in sorted(groups.items()):
            uniq_ids = sorted(set(profile_ids))
            if len(uniq_ids) <= 1:
                continue
            base_name = str(all_rows.get(uniq_ids[0], {}).get("name") or "").strip() or "Profile"
            occupied = {
                _normalize_profile_name(str(row.get("name") or ""))
                for pid, row in all_rows.items()
                if pid not in uniq_ids
            }
            assignments: Dict[str, str] = {}
            next_index = 1
            for profile_id in uniq_ids:
                while True:
                    candidate = f"{base_name} {next_index}"
                    next_index += 1
                    normalized = _normalize_profile_name(candidate)
                    if normalized not in occupied:
                        occupied.add(normalized)
                        assignments[profile_id] = candidate
                        break

            for profile_id, new_name in assignments.items():
                sides = combined.get(profile_id) or {}
                local_row = dict(sides.get("local") or {})
                remote_row = dict(sides.get("remote") or {})
                try:
                    if local_row and str(local_row.get("name") or "").strip() != new_name:
                        local_row["name"] = new_name
                        local_row["updated_at"] = _utc_sql_now()
                        self._upsert_local_profile(local_row)
                    if remote_row and str(remote_row.get("name") or "").strip() != new_name:
                        remote_row["name"] = new_name
                        remote_row["updated_at"] = _utc_sql_now()
                        self._push_remote_profile_metadata(remote_row)
                except Exception as exc:
                    logger.error("Profile name conflict reconcile failed for profile_id=%s: %s", profile_id, exc)
                    had_errors = True
        return had_errors

    # Snapshot helpers
    def _export_local_profile(self, profile_id: str) -> Dict[str, Any]:
        profile_row = self.db.fetch_one(
            "SELECT * FROM profiles WHERE id = %s AND COALESCE(is_deleted, 0) = 0 LIMIT 1",
            (profile_id,),
        )
        if not profile_row:
            raise RuntimeError(f"Local profile not found: {profile_id}")

        table_rows: Dict[str, List[Dict[str, Any]]] = {}
        for table_name in _order_profile_tables(self._discover_profile_tables()):
            rows = self.db.fetch_all(
                f"SELECT * FROM `{table_name}` WHERE profile_id = %s",
                (profile_id,),
            ) or []
            table_rows[table_name] = [_normalize_row(dict(row)) for row in rows if isinstance(row, dict)]

        user_ids = [
            str((row or {}).get("id") or "").strip()
            for row in (table_rows.get("users") or [])
            if str((row or {}).get("id") or "").strip()
        ]
        permissions = self._fetch_rows_by_user_ids("user_permissions", user_ids)
        scopes = self._fetch_rows_by_user_ids("user_access_scopes", user_ids)

        return {
            "schema_version": "profile-sync-v1",
            "profile": _normalize_row(dict(profile_row)),
            "tables": table_rows,
            "related": {
                "user_permissions": permissions,
                "user_access_scopes": scopes,
            },
        }

    def _import_local_snapshot(self, snapshot: Dict[str, Any]) -> None:
        profile_row = _normalize_row(dict(snapshot.get("profile") or {}))
        profile_id = str(profile_row.get("id") or "").strip()
        if not profile_id:
            raise RuntimeError("Snapshot profile id is missing.")
        if self._has_local_tombstone(profile_id):
            raise RuntimeError(f"Profile {profile_id} is tombstoned locally and cannot be auto-restored.")

        tables = snapshot.get("tables") or {}
        related = snapshot.get("related") or {}
        if not isinstance(tables, dict) or not isinstance(related, dict):
            raise RuntimeError("Snapshot payload is invalid.")

        with self.db.transaction_scope() as cursor:
            try:
                cursor.execute("SET FOREIGN_KEY_CHECKS=0")
            except Exception:
                pass
            self._upsert_row_with_cursor(cursor, "profiles", profile_row)
            cursor.execute("SELECT id FROM profiles WHERE id = %s LIMIT 1", (profile_id,))
            if not cursor.fetchone():
                raise RuntimeError(
                    f"Imported profile row is missing after upsert (profile_id={profile_id}). "
                    "This usually means another local profile already uses the same name."
                )
            for table_name in _order_profile_tables(tables.keys()):
                rows = tables.get(table_name) or []
                if not isinstance(rows, list):
                    continue
                for row in rows:
                    if isinstance(row, dict):
                        self._upsert_row_with_cursor(cursor, table_name, _normalize_row(dict(row)))
            for row in related.get("user_permissions") or []:
                if isinstance(row, dict):
                    self._upsert_row_with_cursor(cursor, "user_permissions", _normalize_row(dict(row)))
            for row in related.get("user_access_scopes") or []:
                if isinstance(row, dict):
                    self._upsert_row_with_cursor(cursor, "user_access_scopes", _normalize_row(dict(row)))
            try:
                cursor.execute("SET FOREIGN_KEY_CHECKS=1")
            except Exception:
                pass
            self._ensure_default_local_profile_settings(profile_id, cursor)

    def _export_remote_profile(self, profile_id: str) -> Dict[str, Any]:
        payload = self._request_json("POST", self.EXPORT_ENDPOINT, {"profile_id": profile_id})
        if not isinstance(payload.get("snapshot"), dict):
            raise RuntimeError("Remote profile export returned an invalid snapshot.")
        return payload

    def _import_remote_profile(
        self,
        snapshot: Dict[str, Any],
        transaction_uuid: str,
    ) -> Dict[str, Any]:
        return self._request_json(
            "POST",
            self.IMPORT_ENDPOINT,
            {
                "snapshot": snapshot,
                "device_id": self._get_or_create_device_id(),
                "transaction_uuid": str(transaction_uuid or uuid.uuid4()),
            },
        )

    # Local DB helpers
    def _upsert_local_profile(self, profile_row: Dict[str, Any]) -> None:
        row = dict(profile_row or {})
        row.setdefault("updated_at", _utc_sql_now())
        with self.db.transaction_scope() as cursor:
            self._upsert_row_with_cursor(cursor, "profiles", _normalize_row(row))

    def _push_remote_profile_metadata(self, profile_row: Dict[str, Any]) -> None:
        profile_id = str((profile_row or {}).get("id") or "").strip()
        if not profile_id:
            raise RuntimeError("Remote profile push requires profile id.")
        row = {
            "id": profile_id,
            "name": str((profile_row or {}).get("name") or "").strip(),
            "is_active": int((profile_row or {}).get("is_active") or 1),
            "is_deleted": int((profile_row or {}).get("is_deleted") or 0),
            "updated_at": str((profile_row or {}).get("updated_at") or _utc_sql_now()),
        }
        response = self._request_json(
            "POST",
            self.PUSH_ENDPOINT,
            {
                "transaction_uuid": str(uuid.uuid4()),
                "device_id": self._get_or_create_device_id(),
                "profile_id": profile_id,
                "table_name": "profiles",
                "action_type": "UPDATE",
                "record_id": profile_id,
                "payload_json": {
                    "profile_id": profile_id,
                    "before": None,
                    "after": row,
                    "meta": {
                        "source": "profile_sync_service.metadata_push",
                        "updated_at": row["updated_at"],
                    },
                },
                "db_name": str(self.tenant_manager.load_license().get("db_name", "") or ""),
            },
        )
        if str(response.get("status") or "").lower() != "ok":
            raise RuntimeError(str(response))

    def _discover_profile_tables(self) -> List[str]:
        rows = self.db.fetch_all(
            """
            SELECT DISTINCT TABLE_NAME
            FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE() AND COLUMN_NAME = 'profile_id'
            ORDER BY TABLE_NAME ASC
            """
        ) or []
        tables: List[str] = []
        for row in rows:
            table_name = str((row or {}).get("TABLE_NAME") or "").strip()
            if not table_name or table_name in PROFILE_SYNC_EXCLUDED_TABLES:
                continue
            if table_name == "profiles" or not _is_safe_identifier(table_name):
                continue
            tables.append(table_name)
        return tables

    def _fetch_rows_by_user_ids(self, table_name: str, user_ids: List[str]) -> List[Dict[str, Any]]:
        if not user_ids or not _is_safe_identifier(table_name):
            return []
        placeholders = ", ".join(["%s"] * len(user_ids))
        rows = self.db.fetch_all(
            f"SELECT * FROM `{table_name}` WHERE user_id IN ({placeholders})",
            tuple(user_ids),
        ) or []
        return [_normalize_row(dict(row)) for row in rows if isinstance(row, dict)]

    def _upsert_row_with_cursor(self, cursor, table_name: str, row: Dict[str, Any]) -> None:
        if not _is_safe_identifier(table_name):
            raise RuntimeError(f"Unsafe table name: {table_name}")
        columns = self.db.get_table_columns(table_name) or set()
        if not columns:
            return
        safe_row = {key: value for key, value in row.items() if key in columns and _is_safe_identifier(key)}
        if not safe_row:
            return
        if table_name == "profiles":
            self._upsert_profile_row_with_cursor(cursor, safe_row)
            return
        col_names = list(safe_row.keys())
        placeholders = ", ".join(["%s"] * len(col_names))
        sql = (
            f"INSERT INTO `{table_name}` ("
            + ", ".join(f"`{name}`" for name in col_names)
            + f") VALUES ({placeholders})"
        )
        update_cols = [name for name in col_names if name != "id"]
        if update_cols:
            sql += " ON DUPLICATE KEY UPDATE " + ", ".join(
                f"`{name}` = VALUES(`{name}`)" for name in update_cols
            )
        cursor.execute(sql, tuple(safe_row[name] for name in col_names))

    def _upsert_profile_row_with_cursor(self, cursor, row: Dict[str, Any]) -> None:
        profile_id = str(row.get("id") or "").strip()
        profile_name = str(row.get("name") or "").strip()
        if not profile_id:
            raise RuntimeError("Profile row id is required.")
        if not profile_name:
            raise RuntimeError("Profile row name is required.")

        cursor.execute("SELECT id FROM profiles WHERE id = %s LIMIT 1", (profile_id,))
        existing_by_id = cursor.fetchone()
        if existing_by_id:
            update_cols = [name for name in row.keys() if name != "id"]
            if not update_cols:
                return
            sql = "UPDATE `profiles` SET " + ", ".join(f"`{name}` = %s" for name in update_cols) + " WHERE id = %s"
            cursor.execute(sql, tuple(row[name] for name in update_cols) + (profile_id,))
            return

        cursor.execute("SELECT id FROM profiles WHERE name = %s LIMIT 1", (profile_name,))
        existing_by_name = cursor.fetchone()
        if existing_by_name and str(existing_by_name.get("id") or "").strip() != profile_id:
            raise RuntimeError(
                f"Profile name conflict for '{profile_name}': "
                f"local id={existing_by_name.get('id')} remote id={profile_id}"
            )

        col_names = list(row.keys())
        placeholders = ", ".join(["%s"] * len(col_names))
        sql = (
            "INSERT INTO `profiles` ("
            + ", ".join(f"`{name}`" for name in col_names)
            + f") VALUES ({placeholders})"
        )
        cursor.execute(sql, tuple(row[name] for name in col_names))

    def _ensure_default_local_profile_settings(self, profile_id: str, cursor) -> None:
        default_settings = [
            ("password_enabled", "0", "bool"),
            ("password_hash", "", "str"),
            ("language", "TR", "str"),
        ]
        for setting_key, setting_value, setting_type in default_settings:
            cursor.execute(
                """
                INSERT INTO app_settings (profile_id, setting_key, setting_value, setting_type)
                VALUES (%s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE setting_value = VALUES(setting_value), setting_type = VALUES(setting_type)
                """,
                (profile_id, setting_key, setting_value, setting_type),
            )

    def _record_local_tombstone(
        self,
        profile_id: str,
        profile_name: str,
        transaction_uuid: str,
        cursor,
        device_id: Optional[str] = None,
    ) -> None:
        cursor.execute(
            """
            INSERT INTO deleted_profiles (profile_id, profile_name, transaction_uuid, device_id, deleted_at)
            VALUES (%s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
                profile_name = VALUES(profile_name),
                transaction_uuid = VALUES(transaction_uuid),
                device_id = VALUES(device_id),
                deleted_at = VALUES(deleted_at)
            """,
            (
                profile_id,
                str(profile_name or "").strip(),
                str(transaction_uuid or ""),
                str(device_id or self._safe_device_id()),
                _utc_sql_now(),
            ),
        )

    def _has_local_tombstone(self, profile_id: str) -> bool:
        row = self.db.fetch_one(
            "SELECT profile_id FROM deleted_profiles WHERE profile_id = %s LIMIT 1",
            (str(profile_id or "").strip(),),
        )
        return bool(row)

    def _safe_device_id(self) -> str:
        try:
            return self._get_or_create_device_id()
        except Exception:
            return ""

    def _purge_local_profile_queue(self, profile_id: str, cursor) -> None:
        cursor.execute(
            """
            DELETE FROM local_sync_queue
            WHERE COALESCE(JSON_UNQUOTE(JSON_EXTRACT(payload_json, '$.profile_id')), '') = %s
            """,
            (profile_id,),
        )

    def _clear_local_profile_sync_state(self, profile_id: str, cursor) -> None:
        cursor.execute("DELETE FROM sync_profile_state WHERE profile_id = %s", (profile_id,))

    def _delete_profile_with_cursor(self, profile_id: str, cursor) -> None:
        cursor.execute(
            """
            SELECT DISTINCT c.TABLE_NAME
            FROM INFORMATION_SCHEMA.COLUMNS c
            JOIN INFORMATION_SCHEMA.TABLES t
              ON t.TABLE_SCHEMA = c.TABLE_SCHEMA AND t.TABLE_NAME = c.TABLE_NAME
            WHERE c.TABLE_SCHEMA = DATABASE()
              AND c.COLUMN_NAME = 'profile_id'
              AND t.TABLE_TYPE = 'BASE TABLE'
            """
        )
        rows = cursor.fetchall() or []
        tables = [
            row.get("TABLE_NAME")
            for row in rows
            if isinstance(row, dict) and isinstance(row.get("TABLE_NAME"), str)
        ]
        tables_set = {
            table_name
            for table_name in tables
            if table_name and table_name not in {"profiles", "deleted_profiles"}
        }
        priority = ["financial_transactions", "financial_accounts", "service_jobs", "service_definitions"]
        ordered: List[str] = []
        for table_name in priority:
            if table_name in tables_set:
                ordered.append(table_name)
                tables_set.remove(table_name)
        users_present = "users" in tables_set
        if users_present:
            tables_set.remove("users")
        ordered.extend(sorted(tables_set))
        if users_present:
            ordered.append("users")
        for table_name in ordered:
            if not str(table_name).replace("_", "").isalnum():
                raise RuntimeError(f"Unexpected table name during profile delete: {table_name}")
            cursor.execute(f"DELETE FROM `{table_name}` WHERE profile_id = %s", (profile_id,))
        cursor.execute("DELETE FROM `profiles` WHERE id = %s", (profile_id,))

    # Sync state helpers
    def _set_local_profile_cursor(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
        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)),
        )

    def _get_or_create_device_id(self) -> str:
        row = self.db.fetch_one("SELECT device_id FROM sync_state LIMIT 1")
        device_id = str((row or {}).get("device_id") or "").strip()
        if device_id:
            return device_id

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

    # HTTP helpers
    def _json_ready_payload(self, payload: Optional[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("Profile sync payload normalization failed; falling back to stringified values.")
            return {
                str(key): str(value)
                for key, value in payload.items()
                if isinstance(key, str)
            }

    def _auth_headers(self) -> Dict[str, str]:
        token = self.tenant_manager.get_api_token()
        if not token:
            raise RuntimeError("API token is missing or expired.")
        return {"Authorization": f"Bearer {token}"}

    def _request_json(
        self,
        method: str,
        url: str,
        payload: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        if requests is None:
            raise RuntimeError("requests module is unavailable.")
        is_mac = detect_runtime_platform() == "macos"
        verify_ssl = not is_mac
        if is_mac:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        request_method = str(method or "GET").upper()
        headers = dict(self._auth_headers())
        if is_mac:
            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)
        try:
            if request_method == "GET":
                response = requests.get(
                    url,
                    headers=headers,
                    timeout=safe_timeout,
                    allow_redirects=False,
                    verify=verify_ssl,
                )
            else:
                response = requests.post(
                    url,
                    json=json_payload,
                    headers=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)
                    if request_method == "GET":
                        response = requests.get(
                            redirect_url,
                            headers=headers,
                            timeout=safe_timeout,
                            allow_redirects=False,
                            verify=verify_ssl,
                        )
                    else:
                        response = requests.post(
                            redirect_url,
                            json=json_payload,
                            headers=headers,
                            timeout=safe_timeout,
                            allow_redirects=False,
                            verify=verify_ssl,
                        )
            if response.status_code == 405 and not str(url).endswith("/"):
                retry_url = f"{url}/"
                if request_method == "GET":
                    response = requests.get(
                        retry_url,
                        headers=headers,
                        timeout=safe_timeout,
                        allow_redirects=False,
                        verify=verify_ssl,
                    )
                else:
                    response = requests.post(
                        retry_url,
                        json=json_payload,
                        headers=headers,
                        timeout=safe_timeout,
                        allow_redirects=False,
                        verify=verify_ssl,
                    )
        except Exception as exc:
            raise RuntimeError(str(exc)) from exc

        try:
            data = response.json()
        except Exception:
            data = {"status": "error", "message": response.text}

        if not isinstance(data, dict):
            raise RuntimeError("Server returned an invalid JSON payload.")
        if response.status_code >= 400 or str(data.get("status") or "").lower() not in {"ok", "success"}:
            message = str(data.get("message") or "").strip()
            code = str(data.get("code") or "").strip()
            body_text = str(response.text or "").strip()
            if not message:
                if code:
                    message = code
                elif body_text:
                    message = body_text[:300]
                else:
                    message = f"HTTP {response.status_code} from {url}"
            raise RuntimeError(message)
        return data




