"""
Sync a Raindrop.io collection to DynamoDB, including cover images.

Phases:
  1. Paginate https://api.raindrop.io/rest/v1/raindrops/{id}?page={n}
  2. Download each bookmark's cover image (thread pool) and upload to S3
     under covers/{collection_id}/{_id}.{ext} - skipped if already present
  3. Batch-write records to DynamoDB (keyed on collectionId + _id), storing
     the cover's S3 key/url in the record
  4. Delete stale records (and their cover objects) whose `_id` is absent
     from the latest API response

Environment variables (Lambda):
    RAINDROP_COLLECTION_ID : numeric collection id
    SSM_TOKEN_PARAMETER    : SSM parameter holding the bearer token
    DDB_TABLE_NAME         : DynamoDB table name
    COVER_BUCKET           : S3 bucket for cover images (omit to skip)
    COVER_BASE_URL         : optional public base URL, e.g.
                             https://cdn.example.com - record.coverUrl is
                             derived from it when set

Local testing:
    python raindrop_sync.py [--dry-run] [--no-covers] [--profile X] [--region Y]
"""

import argparse
import concurrent.futures
import json
import logging
import os
import sys
import time
import urllib.error
import urllib.request
from urllib.parse import urlparse

import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)

if len(logging.getLogger().handlers) > 0:
    logging.getLogger().setLevel(logging.INFO)
else:
    logging.basicConfig(level=logging.INFO)

RAINDROP_API_BASE = "https://api.raindrop.io/rest/v1/raindrops"
PAGE_SIZE = 50
MAX_PAGES = 500
MAX_RETRIES = 3
HTTP_TIMEOUT = 30

# Cover images: cap size, map content-types to extensions
MAX_COVER_BYTES = 10 * 1024 * 1024
EXT_BY_TYPE = {
    "image/jpeg": ".jpg",
    "image/png": ".png",
    "image/webp": ".webp",
    "image/gif": ".gif",
    "image/avif": ".avif",
}
USER_AGENT = "raindrop-sync-lambda/1.0"
COVER_WORKERS = 8

# Defaults used only when the corresponding env var is unset (local runs).
DEFAULTS = {
    "RAINDROP_COLLECTION_ID": "12345678",
    "SSM_TOKEN_PARAMETER": "/maceage-blog-raindrop-sync/bearer-token",
    "DDB_TABLE_NAME": "maceage-blog-raindrops",
    "COVER_BUCKET": "blog.maceage.com-raindrop"
}

_ssm = None
_table = None
_s3 = None

def get_config(name: str) -> str:
    """Read a setting from the environment, falling back to a default."""
    return os.environ.get(name, DEFAULTS.get(name, ""))

def init_clients(profile_name=None, region=None):
    """Initialise boto3 clients, optionally with a local profile/region."""
    global _ssm, _table, _s3
    session_kwargs = {}
    if profile_name:
        session_kwargs["profile_name"] = profile_name
    if region:
        session_kwargs["region_name"] = region
    session = boto3.Session(**session_kwargs)
    _ssm = session.client("ssm")
    _s3 = session.client("s3")
    _table = session.resource("dynamodb").Table(get_config("DDB_TABLE_NAME"))

def get_table():
    if _table is None:
        init_clients()
    return _table

def get_bearer_token() -> str:
    """Retrieve the bearer token from SSM Parameter Store."""
    if _ssm is None:
        init_clients()
    response = _ssm.get_parameter(
        Name=get_config("SSM_TOKEN_PARAMETER"),
        WithDecryption=True,
    )
    return response["Parameter"]["Value"]

def fetch_page(collection_id: int, page: int, token: str) -> dict:
    """Fetch a single page of raindrops with retry/backoff."""
    url = f"{RAINDROP_API_BASE}/{collection_id}?page={page}&perpage={PAGE_SIZE}"
    headers = {"Authorization": f"Bearer {token}"}

    last_error = None
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            req = urllib.request.Request(url, headers=headers)
            with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError) as exc:
            last_error = exc
            wait = 2 ** attempt
            logger.warning("Page %d attempt %d failed (%s), retrying in %ds",
                           page, attempt, exc, wait)
            time.sleep(wait)

    raise RuntimeError(f"Failed to fetch page {page} after {MAX_RETRIES} attempts") from last_error

def fetch_all_items(collection_id: int, token: str) -> list:
    """Phase 1: paginate the API into a flat list of items."""
    items = []
    for page in range(0, MAX_PAGES):
        payload = fetch_page(collection_id, page, token)
        page_items = payload.get("items", [])
        if not page_items:
            break
        items.extend(page_items)
        total = payload.get("count")
        fetched = (page * PAGE_SIZE) + len(page_items)
        logger.info("Page %d: %d items (total reported: %s, fetched so far: %d)",
                    page, len(page_items), total, fetched)
        if total is not None and fetched >= total:
            break
    else:
        raise RuntimeError(f"Exceeded MAX_PAGES ({MAX_PAGES}) - aborting sync")
    return items

def pick_cover_url(item: dict):
    """Explicit cover first, then the first image-sized media entry."""
    cover = item.get("cover")
    if isinstance(cover, str) and cover.startswith("https://"):
        return cover
    for media in item.get("media") or []:
        if media.get("type") == "image":
            link = media.get("link")
            if isinstance(link, str) and link.startswith("https://"):
                return link
    return None

def ext_for(content_type: str, url: str) -> str:
    """Extension from content-type, falling back to the URL path suffix."""
    ct = (content_type or "").split(";")[0].strip().lower()
    if ct in EXT_BY_TYPE:
        return EXT_BY_TYPE[ct]
    suffix = os.path.splitext(urlparse(url).path)[1].lower()
    if suffix in (".jpg", ".jpeg", ".png", ".webp", ".gif", ".avif"):
        return ".jpg" if suffix == ".jpeg" else suffix
    return ".jpg"   # raindrop covers are overwhelmingly jpegs

def sniff_image(data: bytes) -> bool:
    """True if the byte stream starts with a known image signature."""
    if len(data) < 12:
        return False
    if data[:3] == b"\xff\xd8\xff":                       # JPEG
        return True
    if data[:8] == b"\x89PNG\r\n\x1a\n":                  # PNG
        return True
    if data[:6] in (b"GIF87a", b"GIF89a"):                # GIF
        return True
    if data[:4] == b"RIFF" and data[8:12] == b"WEBP":      # WebP
        return True
    if data[4:8] == b"ftyp" and data[8:12].startswith(b"avi"):  # AVIF
        return True
    return False

def sync_cover(bucket: str, collection_id: int, item: dict, dry_run: bool):
    """
    Phase 2 worker: download this bookmark's cover and upload it to S3,
    keyed by the bookmark's _id. Returns the S3 key, or None if there is
    no cover, it already exists, or the download failed (non-fatal -
    a missing thumbnail shouldn't sink the whole sync).
    """
    item_id = item.get("_id")
    url = pick_cover_url(item)
    if not item_id or not url:
        return None

    key = None
    last_error = None
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
            with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
                content_type = resp.headers.get("Content-Type", "")
                data = resp.read(MAX_COVER_BYTES + 1)
            if len(data) > MAX_COVER_BYTES:
                logger.warning("Cover for %d exceeds %d bytes; skipping",
                               item_id, MAX_COVER_BYTES)
                return None
            if not data:
                logger.warning("Cover for %d is empty; skipping", item_id)
                return None
            if not sniff_image(data):
                logger.warning(
                    "Cover for %d is not a recognisable image "
                    "(possibly HTML error page, %d bytes); skipping", item_id, len(data))
                return None
            if dry_run:
                return f"(dry-run) bookmark-covers/{collection_id}/{item_id}{ext_for(content_type, url)}"

            key = f"bookmark-covers/{collection_id}/{item_id}{ext_for(content_type, url)}"

            # Skip re-upload if we already have this cover (one-time cost per
            # bookmark; a changed cover on Raindrop will rarely need refreshing,
            # and the existence check saves ~1,400 downloads per nightly sync)
            try:
                _s3.head_object(Bucket=bucket, Key=key)
                return key
            except ClientError as exc:
                if exc.response["Error"]["Code"] not in ("404", "403", "NoSuchKey"):
                    raise

            _s3.put_object(
                Bucket=bucket,
                Key=key,
                Body=data,
                ContentType=content_type.split(";")[0].strip() or "application/octet-stream",
                CacheControl="public, max-age=604800",
            )
            return key

        except Exception as exc:
            last_error = exc
            wait = 2 ** attempt
            logger.warning("Cover fetch for %d attempt %d failed (%s), retrying in %ds",
                           item_id, attempt, exc, wait)
            time.sleep(wait)

    logger.error("Giving up on cover for %d after %d attempts: %s",
                 item_id, MAX_RETRIES, last_error)
    return None

def flatten_item(item: dict, collection_id: int, cover_key, cover_url) -> dict:
    """
    Phase 3: Build the DynamoDB record from a raindrop item.

    The `_id` field becomes the table's sort key `id`, and `collectionId`
    is forced to the collection we're syncing so composite-key queries work.
    """
    record = {
        "collectionId": collection_id,
        "id": item["_id"],
        "title": item["title"],
        "link": item["link"],
        "domain": item["domain"],
        "tags": item["tags"],
        "created": item["created"],
        "raindrop": item,
        "syncedAt": int(time.time()),
    }
    if cover_key:
        record["coverKey"] = cover_key
    if cover_url:
        record["coverUrl"] = cover_url
    return record

def lambda_handler(event, context):
    collection_id = int(get_config("RAINDROP_COLLECTION_ID"))
    token = get_bearer_token()
    table = get_table()
    cover_bucket = get_config("COVER_BUCKET")
    cover_base = get_config("COVER_BASE_URL").rstrip("/")
    dry_run = bool(event.get("dry_run")) if isinstance(event, dict) else False

    # ------------------------------------------------------------------
    # 1 + 2. Fetch items, then fan cover downloads out to a thread pool
    # ------------------------------------------------------------------
    items = fetch_all_items(collection_id, token)

    cover_keys = {}
    if cover_bucket:
        if dry_run:
            logger.info("Dry run: cover fetch simulated")
        with concurrent.futures.ThreadPoolExecutor(max_workers=COVER_WORKERS) as pool:
            futures = {
                pool.submit(sync_cover, cover_bucket, collection_id, item, dry_run): item
                for item in items
            }
            done, failed = 0, 0
            for future in concurrent.futures.as_completed(futures):
                item = futures[future]
                done += 1
                try:
                    key = future.result()
                except Exception as exc:      # workers already retried internally
                    key = None
                    failed += 1
                    logger.error("Cover worker failed for %s: %s", item.get("_id"), exc)
                if key:
                    cover_keys[item["_id"]] = key
                if done % 250 == 0:
                    logger.info("Cover progress: %d/%d done (%d failed)", done, len(items), failed)
        logger.info("Covers resolved: %d of %d items", len(cover_keys), len(items))
    else:
        logger.warning("COVER_BUCKET not set - skipping cover sync")

    # ------------------------------------------------------------------
    # 3. Batch-write records (cover key/url embedded per record)
    # ------------------------------------------------------------------
    seen_ids = set()
    with table.batch_writer() as batch:
        for item in items:
            key = cover_keys.get(item["_id"])
            url = f"{cover_base}/{key}" if key and cover_base else None
            record = flatten_item(item, collection_id, key, url)
            seen_ids.add(record["id"])
            if not dry_run:
                batch.put_item(Item=record)

    # ------------------------------------------------------------------
    # 4. Remove stale records AND their cover objects, matched on `_id`
    # ------------------------------------------------------------------
    deleted = 0
    covers_deleted = 0
    if not dry_run:
        with table.batch_writer() as batch:
            response = table.query(
                KeyConditionExpression=boto3.dynamodb.conditions.Key("collectionId").eq(collection_id),
                ProjectionExpression="id, coverKey",
            )
            while True:
                for row in response.get("Items", []):
                    if row["id"] not in seen_ids:
                        batch.delete_item(Key={
                            "collectionId": collection_id,
                            "id": row["id"],
                        })
                        deleted += 1
                        cover_key = row.get("coverKey")
                        if cover_bucket and cover_key:
                            try:
                                _s3.delete_object(Bucket=cover_bucket, Key=cover_key)
                                covers_deleted += 1
                            except ClientError as exc:
                                logger.warning("Could not delete cover %s: %s", cover_key, exc)

                if "LastEvaluatedKey" not in response:
                    break
                response = table.query(
                    KeyConditionExpression=boto3.dynamodb.conditions.Key("collectionId").eq(collection_id),
                    ProjectionExpression="id, coverKey",
                    ExclusiveStartKey=response["LastEvaluatedKey"],
                )

    result = {
        "collectionId": collection_id,
        "upserted": len(seen_ids),
        "coversStored": len(cover_keys),
        "deleted": deleted,
        "coversDeleted": covers_deleted,
        "dryRun": dry_run,
    }
    logger.info("Sync complete: %s", result)
    return result

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Sync Raindrop.io collection to DynamoDB (with covers)")
    parser.add_argument("--collection-id", type=int, default=None,
                        help="Override RAINDROP_COLLECTION_ID")
    parser.add_argument("--profile", type=str, default=None,
                        help="AWS profile to use (local testing)")
    parser.add_argument("--region", type=str, default=None,
                        help="AWS region to use (local testing)")
    parser.add_argument("--dry-run", action="store_true",
                        help="Fetch and report only; skip DynamoDB writes and deletes")
    parser.add_argument("--no-covers", action="store_true",
                        help="Skip cover image downloads entirely")
    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(levelname)s %(message)s")

    if args.collection_id:
        os.environ["RAINDROP_COLLECTION_ID"] = str(args.collection_id)
    if args.no_covers:
        os.environ.pop("COVER_BUCKET", None)

    # Binds boto3 clients early (honours --profile/--region for local runs)
    init_clients(profile_name=args.profile, region=args.region)

    # event carries the dry_run flag through to the handler
    exit_code = 0
    try:
        result = lambda_handler({"dry_run": args.dry_run}, None)
        print(json.dumps(result, indent=2))
    except Exception:
        logger.exception("Sync failed")
        exit_code = 1

    sys.exit(exit_code)

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Parameters:
  TableName:
    Type: String
    Description: DynamoDB table name

  CollectionId:
    Type: Number
    Description: Raindrop.io collection to sync

  BookmarkCoverBucketName:
    Type: String
    Description: S3 bucket name for bookmark cover images

  TokenParameterName:
    Type: String
    Default: /maceage-blog/raindrop-sync/bearer-token
    Description: SSM parameter name for the bearer token

  ScheduleState:
    Type: String
    Default: DISABLED
    AllowedValues: [ENABLED, DISABLED]
    Description: Whether the nightly sync schedule is active

Resources:
  RaindropTable:
    Type: AWS::DynamoDB::Table
    UpdateReplacePolicy: Retain
    DeletionPolicy: Delete
    Properties:
      TableName: !Ref TableName
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: collectionId
          AttributeType: N
        - AttributeName: id
          AttributeType: N
      KeySchema:
        - AttributeName: collectionId
          KeyType: HASH
        - AttributeName: id
          KeyType: RANGE
      SSESpecification:
        SSEEnabled: true

  RaindropSyncLambda:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: raindrop-sync
      Description: Syncs a Raindrop.io collection to DynamoDB and covers to S3, deleting removed bookmarks and covers
      Runtime: python3.14
      Handler: raindrop_sync.lambda_handler
      Timeout: 900
      MemorySize: 256
      CodeUri: ../raindrop-sync/
      Architectures: [arm64]
      Policies:
        - Statement:
            - Sid: ReadBearerToken
              Effect: Allow
              Action: ssm:GetParameter
              Resource: !Sub
                - arn:${AWS::Partition}:ssm:${AWS::Region}:${AWS::AccountId}:parameter${ParamName}
                - ParamName: !Ref TokenParameterName
        - Statement:
            - Sid: DecryptSecureString
              Effect: Allow
              Action: kms:Decrypt
              Resource: !Sub arn:${AWS::Partition}:kms:${AWS::Region}:${AWS::AccountId}:alias/aws/ssm
        - Statement:
            - Sid: SyncTable
              Effect: Allow
              Action:
                - dynamodb:PutItem
                - dynamodb:DeleteItem
                - dynamodb:Query
                - dynamodb:DescribeTable
                - dynamodb:BatchWriteItem
              Resource: !GetAtt RaindropTable.Arn
        - Statement:
            - Sid: ManageCoverObjects
              Effect: Allow
              Action:
                - s3:PutObject
                - s3:GetObject
                - s3:DeleteObject
              Resource: !Sub arn:${AWS::Partition}:s3:::${BookmarkCoverBucketName}/*
        - Statement:
            - Sid: CreateLogGroup
              Effect: Allow
              Action: logs:CreateLogGroup
              Resource: !Sub arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/raindrop-sync
        - Statement:
            - Sid: PutLogEvents
              Effect: Allow
              Action:
                - logs:CreateLogStream
                - logs:PutLogEvents
              Resource: !Sub arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/raindrop-sync:log-stream:raindrop-sync*
      Environment:
        Variables:
          RAINDROP_COLLECTION_ID: !Ref CollectionId
          SSM_TOKEN_PARAMETER: !Ref TokenParameterName
          DDB_TABLE_NAME: !Ref TableName
      Events:
        NightlySync:
          Type: ScheduleV2
          Properties:
            State: !Ref ScheduleState
            # 17:00 UTC = 04:00/05:00 AEST/AEDT
            ScheduleExpression: cron(0 17 * * ? *)

  SyncIntrospectionPolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      Description: Lets the sync Lambda inspect itself and its nightly schedule
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Sid: GetFunction
            Effect: Allow
            Action: lambda:GetFunction
            Resource: !GetAtt RaindropSyncLambda.Arn
          - Sid: GetSchedule
            Effect: Allow
            Action: scheduler:GetSchedule
            Resource: !GetAtt RaindropSyncLambdaNightlySync.Arn
      Roles:
        - !Ref RaindropSyncLambdaRole

Outputs:
  TableName:
    Description: DynamoDB table holding synced raindrops
    Value: !Ref RaindropTable
  TableArn:
    Description: DynamoDB table ARN
    Value: !GetAtt RaindropTable.Arn
  FunctionArn:
    Description: Sync function ARN
    Value: !GetAtt RaindropSyncLambda.Arn
  TokenParameterName:
    Description: SSM parameter holding the bearer token
    Value: !Ref TokenParameterName

...

- name: Store bearer token
  run: aws ssm put-parameter --name "${{ env.TOKEN_PARAMETER_NAME }}" --value "${{ secrets.RAINDROP_BEARER_TOKEN }}" --type SecureString --overwrite

- name: Install AWS SAM
  uses: aws-actions/setup-sam@v3
  with:
    use-installer: true
    token: ${{ secrets.GITHUB_TOKEN }}

- name: SAM validate
  run: sam validate --template-file ${{ github.workspace }}/deploy/infra-raindrop-sync.yml

- name: SAM build
  run: sam build -t ${{ github.workspace }}/deploy/infra-raindrop-sync.yml

- name: SAM deploy
  run: |
    sam deploy \
      --template-file ${{ github.workspace }}/deploy/infra-raindrop-sync.yml \
      --stack-name MaceageBlogSyncLambdaStack \
      --s3-bucket ${{ env.LAMBDA_S3_BUCKET_NAME }} \
      --s3-prefix raindrop-sync-lambda \
      --force-upload \
      --capabilities CAPABILITY_IAM \
      --no-confirm-changeset \
      --no-fail-on-empty-changeset \
      --parameter-overrides \
        'ParameterKey=TableName,ParameterValue=${{ env.TABLE_NAME }}' \
        'ParameterKey=CollectionId,ParameterValue=${{ secrets.RAINDROP_COLLECTION_ID }}' \
        'ParameterKey=BookmarkCoverBucketName,ParameterValue=${{ env.BOOKMARK_COVER_BUCKET_NAME }}' \
        'ParameterKey=TokenParameterName,ParameterValue=${{ env.TOKEN_PARAMETER_NAME }}' \
        'ParameterKey=ScheduleState,ParameterValue=${{ env.SCHEDULE_STATE }}'

...

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": "lambda:GetFunction",
            "Resource": "arn:aws:lambda:us-east-1:[AccountID]:function:raindrop-sync",
            "Effect": "Allow",
            "Sid": "GetFunction"
        },
        {
            "Action": "scheduler:GetSchedule",
            "Resource": "arn:aws:scheduler:us-east-1:[AccountID]:schedule/default/RaindropSyncLambdaNightlySync",
            "Effect": "Allow",
            "Sid": "GetSchedule"
        }
    ]
}