python/hubspot_sequences.py

"""
HubSpot Sequences → Supabase (incremental since a millisecond cursor)

Sequences are fetched per-user via the HubSpot automation REST API
(no SDK search endpoint available). The cursor filters out sequences
whose updatedAt is at or before since_ms, so only new or changed
sequences are upserted each run.

Usage from orchestrator:
    import hubspot_sequences
    hubspot_sequences.main(since_ms=<int milliseconds since epoch UTC>)

Direct CLI:
    # epoch ms
    python hubspot_sequences.py 1754025600000
    # ISO-8601
    python hubspot_sequences.py 2026-01-01T00:00:00Z
    # Back-compat date (floors to 00:00Z)
    python hubspot_sequences.py 2026-01-01
"""
import os
import re
import time
import logging
import datetime
from typing import Dict, List, Optional, Tuple, Union

import httpx
from dotenv import load_dotenv
from supabase import create_client

from hubspot_utils import parse_ts, try_parse_int
from supabase_utils import insert_into_supabase_table, update_sync_metadata

# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
HUBSPOT_BASE_URL = "https://api.hubapi.com"
HUBSPOT_TIMEOUT = 30.0
HUBSPOT_PAGE_LIMIT = 100
REQUEST_SLEEP_SECONDS = 0.1
MAX_RETRIES = 3

# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logging.basicConfig(
    filename=(
        f"logs/hubspot_sequences_pipeline_"
        f"{datetime.datetime.now().strftime('%Y-%m-%d')}.log"
    ),
    filemode="a",
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

# -----------------------------------------------------------------------------
# Environment
# -----------------------------------------------------------------------------
load_dotenv()
HUBSPOT_TOKEN = os.getenv("HUBSPOT_TOKEN")
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
# Optional bootstrap cursor if orchestrator doesn't provide one
BOOTSTRAP_SINCE_MS_ENV = os.getenv("HUBSPOT_SEQUENCES_SINCE_MS")

if not HUBSPOT_TOKEN:
    raise RuntimeError("HUBSPOT_TOKEN is not set")
if not SUPABASE_URL or not SUPABASE_SERVICE_ROLE_KEY:
    raise RuntimeError("Supabase env vars are not set")

supabase_client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)

# -----------------------------------------------------------------------------
# Time helpers
# -----------------------------------------------------------------------------
def _ensure_utc(dt: datetime.datetime) -> datetime.datetime:
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=datetime.timezone.utc)
    return dt.astimezone(datetime.timezone.utc)


def floor_to_utc_midnight(dt: datetime.datetime) -> datetime.datetime:
    dt = _ensure_utc(dt)
    return dt.replace(hour=0, minute=0, second=0, microsecond=0)


def _parse_iso_like_to_dt(value: str) -> datetime.datetime:
    if isinstance(value, str) and value.endswith("Z"):
        value = value[:-1] + "+00:00"
    dt = datetime.datetime.fromisoformat(value)
    return _ensure_utc(dt)


def to_epoch_ms(dt_or_str: Union[str, datetime.datetime]) -> int:
    if isinstance(dt_or_str, str):
        dt = _parse_iso_like_to_dt(dt_or_str)
    elif isinstance(dt_or_str, datetime.datetime):
        dt = _ensure_utc(dt_or_str)
    else:
        raise TypeError(f"Unsupported type for to_epoch_ms: {type(dt_or_str)}")
    return int(dt.timestamp() * 1000)


def parse_any_ts_ms(value: Optional[Union[str, int, float]]) -> Optional[int]:
    """
    Accepts ms-epoch / sec-epoch / ISO-8601; returns ms since epoch or None.
    """
    if value is None:
        return None
    try:
        v = int(str(value))
        if v < 10_000_000_000_000:  # seconds → ms
            v *= 1000
        return v
    except ValueError:
        pass
    try:
        return to_epoch_ms(str(value))
    except Exception:
        logging.warning("Could not parse timestamp value=%r", value)
        return None


# -----------------------------------------------------------------------------
# HubSpot HTTP helpers
# -----------------------------------------------------------------------------
def _get_hubspot_headers() -> Dict[str, str]:
    return {
        "Authorization": f"Bearer {HUBSPOT_TOKEN}",
        "Content-Type": "application/json",
    }


def hubspot_get(
    client: httpx.Client,
    path: str,
    params: Optional[Dict] = None,
    max_retries: int = MAX_RETRIES,
) -> Optional[Dict]:
    """
    Make a GET request to the HubSpot API with retry and rate-limit handling.

    :param client: The HTTPX client to use.
    :param path: API path (appended to HUBSPOT_BASE_URL).
    :param params: Optional query parameters.
    :param max_retries: Maximum number of retry attempts.
    :return: Parsed JSON response dict, or None if all retries fail.
    """
    url = f"{HUBSPOT_BASE_URL}{path}"

    for attempt in range(1, max_retries + 1):
        try:
            response = client.get(url, headers=_get_hubspot_headers(), params=params)

            if response.status_code == 429:
                retry_after = response.headers.get("Retry-After")
                sleep_seconds = float(retry_after) if retry_after else float(attempt)
                logging.warning(
                    "Rate limited on %s. Sleeping for %s seconds.", url, sleep_seconds
                )
                time.sleep(sleep_seconds)
                continue

            response.raise_for_status()
            return response.json()

        except httpx.HTTPStatusError as exc:
            logging.error(
                "HubSpot request failed [%s] %s: %s",
                exc.response.status_code,
                url,
                exc.response.text,
            )
            if attempt == max_retries:
                raise
            time.sleep(attempt)

        except httpx.RequestError as exc:
            logging.error("HubSpot request error on %s: %s", url, exc)
            if attempt == max_retries:
                raise
            time.sleep(attempt)

    return None


# -----------------------------------------------------------------------------
# Supabase user lookup
# -----------------------------------------------------------------------------
def fetch_hubspot_user_ids() -> List[int]:
    """
    Fetch distinct, non-null user IDs from the hubspot_users table in Supabase.

    :return: List of integer user IDs.
    """
    response = (
        supabase_client.table("hubspot_users")
        .select("user_id")
        .not_.is_("user_id", "null")
        .execute()
    )

    rows = response.data or []
    user_ids: List[int] = []
    seen: set = set()

    for row in rows:
        user_id = try_parse_int(row.get("user_id"))
        if user_id is None or user_id in seen:
            continue
        seen.add(user_id)
        user_ids.append(user_id)

    return user_ids


# -----------------------------------------------------------------------------
# HubSpot sequence fetchers
# -----------------------------------------------------------------------------
def fetch_sequence_summaries_for_user(
    client: httpx.Client,
    user_id: int,
) -> List[Dict]:
    """
    Page through all sequence summaries for a given HubSpot user.

    :param client: The HTTPX client to use.
    :param user_id: HubSpot user ID to scope the request.
    :return: List of raw sequence summary dicts.
    """
    summaries: List[Dict] = []
    after: Optional[str] = None

    while True:
        params: Dict = {"limit": HUBSPOT_PAGE_LIMIT, "userId": user_id}
        if after is not None:
            params["after"] = after

        payload = hubspot_get(client, "/automation/v4/sequences", params=params)
        results = payload.get("results", []) if payload else []
        summaries.extend(results)

        paging = payload.get("paging", {}) if payload else {}
        after = (paging.get("next") or {}).get("after")
        if not after:
            break

        time.sleep(REQUEST_SLEEP_SECONDS)

    return summaries


def fetch_sequence_detail(
    client: httpx.Client,
    sequence_id: int,
    user_id: int,
) -> Optional[Dict]:
    """
    Fetch the full detail payload for a single sequence.

    :param client: The HTTPX client to use.
    :param sequence_id: The sequence ID to fetch.
    :param user_id: HubSpot user ID to scope the request.
    :return: Raw sequence detail dict, or None if the request fails.
    """
    return hubspot_get(
        client,
        f"/automation/v4/sequences/{sequence_id}",
        params={"userId": user_id},
    )


# -----------------------------------------------------------------------------
# Transform helpers
# -----------------------------------------------------------------------------
def transform_sequence(sequence_detail: Dict) -> Dict:
    """
    Flatten a HubSpot sequence detail dict into a Supabase row dict.

    :param sequence_detail: Raw sequence detail from the HubSpot API.
    :return: Flattened sequence record ready for upsert.
    """
    settings = sequence_detail.get("settings") or {}

    return {
        "sequence_id": try_parse_int(sequence_detail.get("id")),
        "name": sequence_detail.get("name"),
        "user_id": try_parse_int(sequence_detail.get("userId")),
        "hubspot_created_at": parse_ts(sequence_detail.get("createdAt")),
        "hubspot_updated_at": parse_ts(sequence_detail.get("updatedAt")),
        "settings_id": try_parse_int(settings.get("id")),
        "eligible_follow_up_days": settings.get("eligibleFollowUpDays"),
        "selling_strategy": settings.get("sellingStrategy"),
        "send_window_start_minute": try_parse_int(settings.get("sendWindowStartMinute")),
        "send_window_end_minute": try_parse_int(settings.get("sendWindowEndMinute")),
        "task_reminder_minute": try_parse_int(settings.get("taskReminderMinute")),
        "individual_task_reminders_enabled": settings.get("individualTaskRemindersEnabled"),
        "settings_hubspot_created_at": parse_ts(settings.get("createdAt")),
        "settings_hubspot_updated_at": parse_ts(settings.get("updatedAt")),
    }


def transform_steps(sequence_detail: Dict) -> List[Dict]:
    """
    Flatten all steps from a sequence detail into a list of Supabase row dicts.

    :param sequence_detail: Raw sequence detail from the HubSpot API.
    :return: List of flattened step records ready for upsert.
    """
    sequence_id = try_parse_int(sequence_detail.get("id"))
    steps: List[Dict] = []

    for step in sequence_detail.get("steps", []):
        task_pattern = step.get("taskPattern") or {}
        email_pattern = step.get("emailPattern") or {}

        steps.append({
            "step_id": try_parse_int(step.get("id")),
            "sequence_id": sequence_id,
            "step_order": try_parse_int(step.get("stepOrder")),
            "delay_millis": try_parse_int(step.get("delayMillis")),
            "action_type": step.get("actionType"),
            "hubspot_created_at": parse_ts(step.get("createdAt")),
            "hubspot_updated_at": parse_ts(step.get("updatedAt")),
            "task_pattern_id": try_parse_int(task_pattern.get("id")),
            "task_type": task_pattern.get("taskType"),
            "task_priority": task_pattern.get("taskPriority"),
            "subject": task_pattern.get("subject"),
            "notes": task_pattern.get("notes"),
            "task_pattern_hubspot_created_at": parse_ts(task_pattern.get("createdAt")),
            "task_pattern_hubspot_updated_at": parse_ts(task_pattern.get("updatedAt")),
            "email_pattern_id": try_parse_int(email_pattern.get("id")),
            "email_template_id": try_parse_int(email_pattern.get("templateId")),
            "email_pattern_hubspot_created_at": parse_ts(email_pattern.get("createdAt")),
            "email_pattern_hubspot_updated_at": parse_ts(email_pattern.get("updatedAt")),
            "thread_email_to_step_order": try_parse_int(
                email_pattern.get("threadEmailToStepOrder")
            ),
        })

    return steps


def transform_dependencies(sequence_detail: Dict) -> List[Dict]:
    """
    Flatten all dependencies from a sequence detail into a list of Supabase row dicts.

    :param sequence_detail: Raw sequence detail from the HubSpot API.
    :return: List of flattened dependency records ready for upsert.
    """
    sequence_id = try_parse_int(sequence_detail.get("id"))
    dependencies: List[Dict] = []

    for dependency in sequence_detail.get("dependencies", []):
        dependencies.append({
            "dependency_id": try_parse_int(dependency.get("id")),
            "sequence_id": sequence_id,
            "dependency_type": dependency.get("dependencyType"),
            "required_by_sequence_step_id": try_parse_int(
                dependency.get("requiredBySequenceStepId")
            ),
            "relies_on_sequence_step_id": try_parse_int(
                dependency.get("reliesOnSequenceStepId")
            ),
            "required_by_step_order": try_parse_int(dependency.get("requiredByStepOrder")),
            "relies_on_step_order": try_parse_int(dependency.get("reliesOnStepOrder")),
            "hubspot_created_at": parse_ts(dependency.get("createdAt")),
            "hubspot_updated_at": parse_ts(dependency.get("updatedAt")),
        })

    return dependencies


# -----------------------------------------------------------------------------
# Fetch + transform all sequences (with cursor filtering)
# -----------------------------------------------------------------------------
def fetch_all_sequence_details(
    since_ms: int,
) -> Tuple[List[Dict], List[Dict], List[Dict], Optional[int]]:
    """
    Fetch all sequence details from HubSpot for every user in Supabase,
    filtering to sequences updated after since_ms.

    The HubSpot sequences API does not support server-side timestamp filtering,
    so summaries are fetched per user and filtered client-side by updatedAt.

    :param since_ms: Only include sequences whose updatedAt > since_ms (epoch ms).
    :return: (sequences, steps, dependencies, max_updated_at_ms) where
             max_updated_at_ms is the highest updatedAt seen, or None.
    """
    user_ids = fetch_hubspot_user_ids()
    logging.info("Fetched %d HubSpot user IDs from Supabase.", len(user_ids))

    if not user_ids:
        return [], [], [], None

    sequences: List[Dict] = []
    steps: List[Dict] = []
    dependencies: List[Dict] = []
    processed_sequence_ids: set = set()
    max_updated_at_ms: Optional[int] = None

    with httpx.Client(timeout=HUBSPOT_TIMEOUT) as client:
        for user_id in user_ids:
            logging.info("Fetching sequence summaries for user_id=%s", user_id)
            summaries = fetch_sequence_summaries_for_user(client, user_id)
            logging.info(
                "Fetched %d sequence summaries for user_id=%s", len(summaries), user_id
            )

            for summary in summaries:
                sequence_id = try_parse_int(summary.get("id"))
                if sequence_id is None or sequence_id in processed_sequence_ids:
                    continue

                # Client-side cursor filter: skip sequences not updated since cursor
                updated_at_ms = parse_any_ts_ms(summary.get("updatedAt"))
                if updated_at_ms is not None and updated_at_ms <= since_ms:
                    continue

                logging.info("Fetching sequence detail for sequence_id=%s", sequence_id)
                detail = fetch_sequence_detail(client, sequence_id, user_id)
                if not detail:
                    logging.warning(
                        "No detail payload returned for sequence_id=%s", sequence_id
                    )
                    continue

                processed_sequence_ids.add(sequence_id)

                # Track max updatedAt for cursor advancement
                if updated_at_ms is not None and (
                    max_updated_at_ms is None or updated_at_ms > max_updated_at_ms
                ):
                    max_updated_at_ms = updated_at_ms

                sequences.append(transform_sequence(detail))
                steps.extend(transform_steps(detail))
                dependencies.extend(transform_dependencies(detail))

                time.sleep(REQUEST_SLEEP_SECONDS)

    return sequences, steps, dependencies, max_updated_at_ms


# -----------------------------------------------------------------------------
# Upsert
# -----------------------------------------------------------------------------
def upsert_sequences(
    sequences: List[Dict],
    steps: List[Dict],
    dependencies: List[Dict],
) -> None:
    if sequences:
        insert_into_supabase_table(supabase_client, "hubspot_sequences", sequences)
        print(f"Upserted {len(sequences)} sequences.")

    if steps:
        insert_into_supabase_table(supabase_client, "hubspot_sequence_steps", steps)
        print(f"Upserted {len(steps)} sequence steps.")

    if dependencies:
        insert_into_supabase_table(
            supabase_client, "hubspot_sequence_dependencies", dependencies
        )
        print(f"Upserted {len(dependencies)} sequence dependencies.")


# -----------------------------------------------------------------------------
# Main (timestamp cursor)
# -----------------------------------------------------------------------------
def main(since_ms: Optional[int] = None) -> None:
    """
    Orchestrates:
      1) Fetch sequence details per user, filtered by since_ms cursor
      2) Upsert sequences, steps, and dependencies into Supabase
      3) Update sync metadata for all three tables
    """
    # Resolve since_ms
    if since_ms is None and BOOTSTRAP_SINCE_MS_ENV:
        try:
            since_ms = int(BOOTSTRAP_SINCE_MS_ENV)
        except ValueError:
            raise RuntimeError("HUBSPOT_SEQUENCES_SINCE_MS must be an integer (ms) if set.")

    if since_ms is None:
        # Default: today@00:00:00Z for first run
        today0 = floor_to_utc_midnight(datetime.datetime.now(datetime.timezone.utc))
        since_ms = to_epoch_ms(today0)

    print(f"Fetching sequences updated after {since_ms} ms ...")
    logging.info("Fetching sequences updated after %d ms.", since_ms)

    sequences, steps, dependencies, max_updated_at_ms = fetch_all_sequence_details(since_ms)

    now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()

    if not sequences:
        print("No sequences found beyond the cursor. Updating sync metadata and exiting.")
        logging.warning("No sequences returned beyond cursor %d.", since_ms)
        update_sync_metadata(supabase_client, "hubspot_sequences", now_iso)
        update_sync_metadata(supabase_client, "hubspot_sequence_steps", now_iso)
        update_sync_metadata(supabase_client, "hubspot_sequence_dependencies", now_iso)
        return

    print(f"Fetched {len(sequences)} sequences from HubSpot.")
    print(f"Fetched {len(steps)} sequence steps from HubSpot.")
    print(f"Fetched {len(dependencies)} sequence dependencies from HubSpot.")

    upsert_sequences(sequences, steps, dependencies)

    new_cursor_ms = max_updated_at_ms if max_updated_at_ms is not None else since_ms

    update_sync_metadata(supabase_client, "hubspot_sequences", now_iso)
    update_sync_metadata(supabase_client, "hubspot_sequence_steps", now_iso)
    update_sync_metadata(supabase_client, "hubspot_sequence_dependencies", now_iso)

    print(f"Sequences sync complete. Advanced cursor to {new_cursor_ms}.")
    logging.info("Sequences sync complete. Advanced cursor to %d.", new_cursor_ms)


# -----------------------------------------------------------------------------
# CLI
# -----------------------------------------------------------------------------
def _parse_cli_arg_to_ms(arg: str) -> int:
    """
    Accept:
      - integer epoch ms
      - ISO-8601 (Z or offset)
      - YYYY-MM-DD (floors to 00:00Z)
    """
    # epoch ms or seconds
    if re.fullmatch(r"\d{10,13}", arg):
        v = int(arg)
        if v < 10_000_000_000_000:  # seconds -> ms
            v *= 1000
        return v

    # YYYY-MM-DD
    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", arg):
        d = datetime.datetime.strptime(arg, "%Y-%m-%d").replace(tzinfo=datetime.timezone.utc)
        return to_epoch_ms(floor_to_utc_midnight(d))

    # ISO-8601
    return to_epoch_ms(arg)


if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1:
        try:
            since = _parse_cli_arg_to_ms(sys.argv[1])
        except Exception as e:
            print(
                f"Invalid timestamp. Provide epoch ms, ISO-8601, or YYYY-MM-DD. Error: {e}"
            )
            sys.exit(1)
        main(since_ms=since)
    else:
        main()