#!/usr/bin/env python3
# Copyright (c) 2026 Asteria Labs. All rights reserved.
"""Prepare a Google Workspace OAuth connector from authenticated Google Cloud Shell.

Standalone, standard-library only. Review the plan before approving changes.
The OAuth consent screen and Web client are completed in Google's console.
"""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
import time
from pathlib import Path
from urllib.parse import urlsplit

# Canonical scopes: backend/app/services/connectors/scopes.py. CI verifies parity.
BASE_SCOPES = ["openid", "email"]
PREFIX = "https://www.googleapis.com/auth/"
SERVICE_SCOPES = {
    "calendar": [PREFIX + "calendar.events", PREFIX + "calendar.calendarlist.readonly"],
    "drive": [PREFIX + "drive.readonly"],
    "drive_file": [PREFIX + "drive.file"],
    "gmail": [PREFIX + "gmail.modify"],
    "chat": [
        PREFIX + scope
        for scope in (
            "chat.users.readstate.readonly",
            "chat.memberships.readonly",
            "chat.messages.readonly",
            "chat.messages.create",
            "chat.spaces.create",
            "chat.spaces.readonly",
        )
    ],
    "docs": [PREFIX + "drive"],
    "sheets": [PREFIX + "drive"],
    "slides": [PREFIX + "drive"],
    "drive_write": [PREFIX + "drive"],
    "directory": [
        PREFIX + "admin.directory.group.readonly",
        PREFIX + "cloud-identity.groups.readonly",
    ],
}
# Full Drive scopes make all four editing services available at consent time.
EDITORS = {"docs", "sheets", "slides", "drive_write"}
SERVICE_APIS = {
    "calendar": {"calendar-json.googleapis.com"},
    "drive": {"drive.googleapis.com"},
    "drive_file": {"drive.googleapis.com", "picker.googleapis.com", "sheets.googleapis.com"},
    "gmail": {"gmail.googleapis.com"},
    "chat": {"chat.googleapis.com"},
    "docs": {"drive.googleapis.com", "docs.googleapis.com"},
    "sheets": {"drive.googleapis.com", "sheets.googleapis.com"},
    "slides": {"drive.googleapis.com", "slides.googleapis.com"},
    "drive_write": {"drive.googleapis.com"},
    "directory": {"admin.googleapis.com", "cloudidentity.googleapis.com"},
}
DEFAULT_REDIRECT = "https://app.asteria-labs.com/api/connectors/callback"


def selection(raw: str) -> list[str]:
    services = list(dict.fromkeys(part.strip() for part in raw.split(",")))
    unknown = set(services) - SERVICE_SCOPES.keys()
    if unknown:
        raise ValueError("Unknown services: " + ", ".join(sorted(unknown)))
    return services


def required_apis(services: list[str]) -> list[str]:
    effective = set(services)
    if effective & EDITORS:
        effective |= EDITORS
    return sorted(set().union(*(SERVICE_APIS[service] for service in effective)))


def gcloud(*args: str) -> str:
    result = subprocess.run(["gcloud", *args], capture_output=True, text=True, check=False)
    if result.returncode:
        # gcloud is never passed customer secrets. Preserve its actionable IAM errors.
        raise ValueError(f"gcloud {' '.join(args)} failed:\n{result.stderr.strip()}")
    return result.stdout.strip()


def describe_created_resource(*args: str) -> dict:
    """IAM reads can lag a successful create by a minute or more."""
    delays = iter((2, 4, 8, 16, 30, 30, 30))
    while True:
        try:
            return json.loads(gcloud(*args))
        except ValueError as exc:
            if not re.search(r"\bNOT_FOUND\b", str(exc)):
                raise
            delay = next(delays, None)
            if delay is None:
                raise ValueError(
                    f"Created resource is not visible after bounded retries. Rerun the same setup to resume.\n{exc}"
                ) from exc
            print(f"Waiting {delay}s for the created resource to become visible...", flush=True)
            time.sleep(delay)


def validate_credentials(path: Path, project: str, redirect: str) -> tuple[str, str]:
    try:
        data = json.loads(path.read_text())
    except (OSError, ValueError):
        raise ValueError("Cannot read the downloaded OAuth client JSON file.") from None
    web = data.get("web") if isinstance(data, dict) else None
    if not isinstance(web, dict):
        raise ValueError(
            "Download a Web application OAuth client, not a desktop or service-account key."
        )
    if web.get("project_id") != project:
        raise ValueError("The OAuth client JSON belongs to a different project.")
    redirects = web.get("redirect_uris")
    if not isinstance(redirects, list) or redirect not in redirects:
        raise ValueError(
            "The OAuth client JSON does not contain the exact callback URL. Fix it and download again."
        )
    client = web.get("client_id")
    secret = web.get("client_secret")
    if not isinstance(client, str) or not re.fullmatch(
        r"[A-Za-z0-9._-]+\.apps\.googleusercontent\.com", client
    ):
        raise ValueError("The OAuth client ID is missing or malformed.")
    if not isinstance(secret, str) or not secret.strip() or any(ord(c) < 32 for c in secret):
        raise ValueError("The OAuth client secret is missing or malformed.")
    return client, secret


def instructions(project: str, services: list[str], redirect: str) -> None:
    root = "https://console.cloud.google.com"
    scopes = sorted(set(BASE_SCOPES).union(*(SERVICE_SCOPES[s] for s in services)))
    origins = (
        "Add each application origin under Authorized JavaScript origins for Google Picker."
        if "drive_file" in services
        else "No JavaScript origin is needed for this server-side connector."
    )
    print(f"""
Next: finish the OAuth application in project {project}.
1. Branding: {root}/auth/branding?project={project}
   Get started if prompted. Name: Asteria Cloud connector. Use your own support
   and contact email addresses. In App information, upload the logo downloaded from:
   https://app.asteria-labs.com/pwa-512x512.png
   In App domain, use these hosted-service values:
   Application home page: https://asteria-labs.com/
   Privacy policy: https://app.asteria-labs.com/privacy
   Terms of service: https://app.asteria-labs.com/terms
   Authorized domains, if requested: asteria-labs.com (no scheme or path).
   For self-hosting, use your operator's applicable pages and branding instead.
   If External requires domain ownership verification, contact Asteria Labs;
   customers cannot verify ownership of asteria-labs.com. Save Branding.
2. Audience: {root}/auth/audience?project={project}
   Choose Internal for members of your Google Workspace organization.
   If Internal is unavailable, check that this project belongs to your Workspace
   organization. Do not silently choose External: Testing connections expire
   after seven days; restricted scopes can require verification and assessment.
3. Data Access: {root}/auth/scopes?project={project}
   Add or remove scopes, then paste these exact scope names (deduplicated):""")
    print("\n".join("   " + scope for scope in scopes))
    if "drive_file" in services:
        print("Drive import uses a separate personal consent, not a combined service grant.")
        print(
            "Create a browser API key restricted to Google Picker API; allow the application referrers and https://docs.google.com/*."
        )
        print(
            "The deployment operator must configure GOOGLE_PICKER_API_KEY, GOOGLE_PICKER_CLIENT_ID and GOOGLE_PICKER_PROJECT_NUMBER for this same OAuth client and project."
        )

    print(f"""4. Clients: {root}/auth/clients?project={project}
   Create client > Web application. Name: Asteria Cloud connector.
   Authorized redirect URI (copy exactly):
   {redirect}
   {origins}
   Copy the client ID and secret now; keep them in your secret manager.
   Download the client JSON if offered, then upload it into Cloud Shell to run
   this script again with --credentials PATH and --verify-only.
   Repeat your original --services and --redirect-uri options if customized.
5. As an organization administrator, open the Asteria Cloud admin console:
   https://console.asteria-labs.com/connectors
   Select Google. Save client ID and client secret,
   then enable Google Workspace. An organization registration takes precedence
   over the platform application for new connections.
6. In Profile > Connections, connect Google as a pilot member and select the
   intended services. Complete consent, then try reading a known file/calendar
   or message. This is the real OAuth token-exchange and service-access test.
   Do not call setup complete until it succeeds.

If Workspace blocks the app, a Workspace administrator must review the OAuth
client in admin.google.com > Security > Access and data control > API controls.
Approve only the intended organizational units and Google data access.
""")
    if "chat" in services:
        print(f"""Chat also needs API configuration for sending messages/creating spaces:
{root}/apis/api/chat.googleapis.com/hangouts-chat?project={project}
For a dedicated connector project, set name and description. Avatar URL:
https://app.asteria-labs.com/pwa-512x512.png
Turn Interactive features OFF, then save. Chat apps Owner permissions are needed.
Use a separate project if one already hosts an interactive Chat app; do not change
that app. No bot appears in Google Chat with Interactive features OFF.
Test from Asteria Cloud: connect Google with Chat selected, ask Asteria to list
spaces/read a known thread, then send a test message to a confirmed test space.
It sends as your connected Google account.
To deploy a project-owned agent that replies to DMs or @mentions in Google Chat,
follow the separate interactive-app steps in the guide below.
""")
    if "directory" in services:
        print(
            "Directory is included in the default all-services setup. Keep it selected\n"
            "when members connect in Profile > Connections, then test a group-shared file.\n"
            "For direct-group fallback, a Workspace admin or Groups Reader must connect\n"
            "with Directory and be selected under Directory account for collection access\n"
            "in the admin console's Connectors page. Scopes do not grant admin privileges.\n"
            "Nested-group lookup depends on the account's rights and Workspace edition.\n"
        )
    if "gmail" in services:
        print(f"""For incoming-mail workflows, complete Gmail Pub/Sub setup separately:
{root}/cloudpubsub/topic/list?project={project}
Enable the Pub/Sub API, create a topic in this OAuth client's project, and grant
Pub/Sub Publisher on it to gmail-api-push@system.gserviceaccount.com.
In the admin console > Connectors > Google > Update the application, save its full
name in Gmail Pub/Sub topic, for example projects/{project}/topics/asteria-gmail.
Then copy the Gmail push endpoint shown on the Google card and create a push
subscription on that topic using the full endpoint (which contains a secret).
This script does not create Pub/Sub resources or change IAM policies.
Setup and Domain Restricted Sharing guidance:
https://docs.asteria-labs.com/admin-guide/connectors#workflows-that-trigger-on-incoming-mail
Test by enabling an incoming-mail workflow, sending a matching message, and
checking its run history. Without push, the fallback check runs every 20 minutes.
Ordinary Gmail reading/sending does not need a topic.
""")
    print(
        "Guide: https://docs.asteria-labs.com/admin-guide/connectors-google\n"
        "Troubleshooting: redirect_uri_mismatch means the Web client's callback must\n"
        "match exactly. access_denied may mean Workspace policy or the wrong audience.\n"
        "API-disabled errors name the API to enable in this same project.\n"
        "Never paste secrets or downloaded client JSON into a support ticket."
    )


EVENTS_APIS = {"workspaceevents.googleapis.com", "pubsub.googleapis.com", "iam.googleapis.com"}


def events_resources(project: str, callback: str) -> dict[str, str]:
    """Derive stable per-organization names from the application's exact callback."""
    url = urlsplit(callback)
    match = re.fullmatch(
        r"/api/connectors/google/workspace-events/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})",
        url.path,
    )
    if (
        url.scheme != "https"
        or not url.hostname
        or url.username
        or url.password
        or url.query
        or url.fragment
        or not match
        or any(c.isspace() or ord(c) < 32 for c in callback)
    ):
        raise ValueError("Use the exact HTTPS Workspace Events callback from Asteria Cloud.")
    name = "asteria-events-" + match[1]
    return {
        "project_id": project,
        "topic": f"projects/{project}/topics/{name}",
        "push_subscription": f"projects/{project}/subscriptions/{name}-push",
        "push_service_account": f"asteria-events-push@{project}.iam.gserviceaccount.com",
    }


def provision_events(project: str, callback: str, *, verify_only: bool) -> None:
    """Reconcile dedicated Pub/Sub resources; never repoint an existing subscription."""
    config = events_resources(project, callback)
    project_flag = "--project=" + project
    number = gcloud("projects", "describe", project, "--format=value(projectNumber)")
    if not re.fullmatch(r"[0-9]+", number):
        raise ValueError("Google did not return a numeric project number.")
    agent = f"serviceAccount:service-{number}@gcp-sa-pubsub.iam.gserviceaccount.com"
    email = config["push_service_account"]
    topic = config["topic"]
    subscription = config["push_subscription"]
    dead_topic = topic + "-dead-letter"
    dead_subscription = subscription + "-dead-letter"
    if not verify_only:
        gcloud(
            "beta",
            "services",
            "identity",
            "create",
            "--service=pubsub.googleapis.com",
            project_flag,
            "--quiet",
        )

    def ensure(group: tuple[str, str], name: str, key: str, *flags: str) -> dict:
        rows = json.loads(gcloud(*group, "list", project_flag, "--format=json"))
        existing = next((row for row in rows if row.get(key) == name), None)
        if existing is None:
            if verify_only:
                raise ValueError(
                    f"Missing Workspace Events resource: {name}. Rerun without --verify-only."
                )
            create_name = "asteria-events-push" if key == "email" else name
            try:
                gcloud(*group, "create", create_name, *flags, project_flag, "--quiet")
            except ValueError as exc:
                # A stale list or concurrent setup can miss an existing resource.
                # Read it and validate its configuration instead of recreating it.
                if not re.search(r"\bALREADY_EXISTS\b", str(exc)):
                    raise
            existing = describe_created_resource(
                *group, "describe", name, project_flag, "--format=json"
            )
        return existing

    def binding(group: tuple[str, str], name: str, member: str, role: str) -> None:
        policy = json.loads(gcloud(*group, "get-iam-policy", name, project_flag, "--format=json"))
        if any(
            b.get("role") == role and member in b.get("members", []) and not b.get("condition")
            for b in policy.get("bindings", [])
        ):
            return
        if verify_only:
            raise ValueError(f"Missing {role} for {member} on {name}.")
        gcloud(
            *group,
            "add-iam-policy-binding",
            name,
            "--member=" + member,
            "--role=" + role,
            *(["--condition=None"] if group == ("iam", "service-accounts") else []),
            project_flag,
            "--quiet",
        )

    account = ensure(("iam", "service-accounts"), email, "email")
    if account.get("disabled"):
        raise ValueError(f"Push service account is disabled: {email}")
    # Leave existing topic storage/residency policies intact, including inherited constraints.
    for name in (topic, dead_topic):
        ensure(("pubsub", "topics"), name, "name")
    binding(
        ("pubsub", "topics"),
        topic,
        "serviceAccount:drive-api-event-push@system.gserviceaccount.com",
        "roles/pubsub.publisher",
    )
    binding(("iam", "service-accounts"), email, agent, "roles/iam.serviceAccountTokenCreator")
    binding(("pubsub", "topics"), dead_topic, agent, "roles/pubsub.publisher")

    for name, target, push in ((dead_subscription, dead_topic, False), (subscription, topic, True)):
        flags = [
            "--topic=" + target,
            "--message-retention-duration=7d",
            "--expiration-period=never",
        ]
        if push:
            flags += [
                "--push-endpoint=" + callback,
                "--push-auth-service-account=" + email,
                "--push-auth-token-audience=" + callback,
                "--min-retry-delay=10s",
                "--max-retry-delay=600s",
                "--dead-letter-topic=" + dead_topic,
                "--max-delivery-attempts=20",
            ]
        row = ensure(("pubsub", "subscriptions"), name, "name", *flags)
        expected_push = (
            {
                "pushEndpoint": callback,
                "oidcToken": {"serviceAccountEmail": email, "audience": callback},
            }
            if push
            else {}
        )
        actual_push = row.get("pushConfig", {})
        # API may return empty attributes and its default wrapped-delivery marker.
        actual_push = {
            k: v for k, v in actual_push.items() if k not in {"attributes", "pubsubWrapper"}
        }
        if (
            row.get("topic") != target
            or actual_push != expected_push
            or row.get("filter")
            or row.get("messageTransforms")
            or row.get("bigqueryConfig")
            or row.get("cloudStorageConfig")
            or row.get("expirationPolicy", {}).get("ttl")
            or row.get("messageRetentionDuration") != "604800s"
            or (
                push
                and (
                    row.get("deadLetterPolicy")
                    != {"deadLetterTopic": dead_topic, "maxDeliveryAttempts": 20}
                    or row.get("retryPolicy") != {"minimumBackoff": "10s", "maximumBackoff": "600s"}
                )
            )
        ):
            raise ValueError(
                f"Existing subscription configuration conflicts: {name}. Inspect it before changing it; the script will not retarget or overwrite it."
            )
    binding(("pubsub", "subscriptions"), subscription, agent, "roles/pubsub.subscriber")
    print("\nWorkspace Events resources checked. Copy into Asteria Cloud > Connectors > Google:")
    print(json.dumps(config, indent=2))
    print(
        f"Callback and OIDC audience: {callback}\nDead-letter inspection subscription: {dead_subscription}"
    )
    print(
        "Drive Developer Preview enrollment and real authenticated delivery still require validation. Monitor delivery errors and the dead-letter backlog; this script does not prove delivery."
    )


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--project", help="existing project ID, or new ID with --create-project")
    parser.add_argument(
        "--create-project", action="store_true", help="explicitly create the project"
    )
    parser.add_argument(
        "--organization", help="numeric Workspace organization ID for a new project"
    )
    parser.add_argument("--services", help="comma-separated service keys; default: all")
    parser.add_argument("--redirect-uri", default=DEFAULT_REDIRECT)
    parser.add_argument(
        "--credentials", type=Path, help="downloaded Web OAuth client JSON; checked locally"
    )
    parser.add_argument(
        "--verify-only", action="store_true", help="read-only API/configuration checks"
    )
    parser.add_argument("--list-services", action="store_true")
    parser.add_argument("--yes", action="store_true", help="approve the displayed cloud changes")
    parser.add_argument(
        "--workspace-events-callback",
        help="provision Workspace Events Pub/Sub using the exact generated HTTPS callback",
    )
    args = parser.parse_args()
    if args.list_services:
        print("\n".join(SERVICE_SCOPES))
        return
    if not shutil.which("gcloud"):
        raise ValueError("Open authenticated Google Cloud Shell: https://shell.cloud.google.com")
    project = args.project or input("Google Cloud project ID (not its display name): ").strip()
    if not re.fullmatch(r"[a-z][a-z0-9-]{4,28}[a-z0-9]", project):
        raise ValueError(
            "Project ID must be 6-30 lowercase letters, digits or hyphens, starting with a letter."
        )
    if args.create_project and (args.verify_only or args.credentials):
        raise ValueError(
            "Create the project first; verification/credentials require an existing project."
        )
    if args.organization and (
        not args.create_project or not re.fullmatch(r"[0-9]+", args.organization)
    ):
        raise ValueError("--organization requires --create-project and a numeric organization ID.")
    if args.create_project and not args.organization:
        raise ValueError(
            "Use --organization NUMBER so the new project can have an Internal audience."
        )
    url = urlsplit(args.redirect_uri)
    if (
        url.scheme != "https"
        or not url.hostname
        or url.username
        or url.password
        or url.fragment
        or url.query
        or any(c.isspace() or ord(c) < 32 for c in args.redirect_uri)
    ):
        raise ValueError(
            "The callback must be an HTTPS URL without credentials, query or fragment."
        )
    print("Default services include Directory for group access to shared files.")
    raw = args.services
    if raw is None:
        # Picker requires separate browser/deployment setup and stays opt-in.
        default = ",".join(service for service in SERVICE_SCOPES if service != "drive_file")
        raw = (
            default
            if args.yes or args.verify_only
            else input(f"Services [{default}]: ").strip() or default
        )
    services = selection(raw)
    apis = required_apis(services)
    events = None
    if args.workspace_events_callback:
        if not ({"drive"} | EDITORS) & set(services):
            raise ValueError(
                "Workspace Events requires selecting drive or a Drive editing service."
            )
        events = events_resources(project, args.workspace_events_callback)
        apis = sorted(set(apis) | EVENTS_APIS)
    credentials = (
        validate_credentials(args.credentials, project, args.redirect_uri)
        if args.credentials
        else None
    )
    account = gcloud("auth", "list", "--filter=status:ACTIVE", "--format=value(account)")
    if not account:
        raise ValueError("No active Google account. Sign into Google Cloud Shell first.")
    if not args.create_project:
        gcloud("projects", "describe", project, "--format=value(projectId)")
    print(f"Account: {account}\nProject: {project}\nCallback: {args.redirect_uri}")
    print("Selected services: " + ", ".join(services))
    print("APIs to check" + ("/enable:" if not args.verify_only else ":"))
    print("\n".join("  " + api for api in apis))
    if events:
        print(
            "Workspace Events plan: create/reuse these resources and their dead-letter topic/subscription:"
        )
        print(json.dumps(events, indent=2))
        print(f"Push endpoint and OIDC audience: {args.workspace_events_callback}")
        print(
            "Grant Drive topic publisher, Pub/Sub token creator on the push account, and dead-letter publisher/subscriber. Retain messages 7 days; retry 10-600 seconds; forward after approximately 20 attempts. Subscriptions never expire from inactivity."
        )
        print(
            "Existing topic residency policies remain unchanged. The operator needs iam.serviceAccounts.actAs and resource/IAM management permissions. No billing, service-account keys or user consent is created."
        )
    else:
        print(
            "No billing account, IAM grants, service-account keys or user consent will be created."
        )
    if args.create_project:
        print(f"Will create project {project} in organization {args.organization}.")
    if not args.verify_only:
        if not args.yes and input("Apply this plan? [y/N] ").strip().lower() not in {"y", "yes"}:
            print("Cancelled. No changes made.")
            return
        if args.create_project:
            gcloud(
                "projects",
                "create",
                project,
                "--organization=" + args.organization,
                "--name=Asteria Cloud connector",
                "--quiet",
            )
        gcloud("services", "enable", *apis, "--project=" + project, "--quiet")
    enabled = set(
        gcloud(
            "services", "list", "--enabled", "--project=" + project, "--format=value(config.name)"
        ).splitlines()
    )
    missing = set(apis) - enabled
    if missing:
        raise ValueError(
            "APIs still missing: "
            + ", ".join(sorted(missing))
            + ". Rerun for this same project after checking permissions/propagation."
        )
    print("Required APIs are enabled.")
    if events:
        provision_events(project, args.workspace_events_callback, verify_only=args.verify_only)
    instructions(project, services, args.redirect_uri)
    if credentials:
        print(
            "\nDownloaded client configuration matches the project and callback.\n"
            "This does NOT verify the secret with Google, the audience, consent or user access.\n"
            "Complete the pilot connection described above. Copy into the admin console > Connectors > Google:"
        )
        print(f"Client ID: {credentials[0]}\nClient secret: {credentials[1]}")
        print("Save securely, then remove the downloaded JSON and any Cloud Shell copy.")
    else:
        print("\nOAuth client not checked. Follow the console steps above; setup is unfinished.")


if __name__ == "__main__":
    try:
        main()
    except (ValueError, EOFError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        print("\nCancelled. Rerun with the same project to resume.", file=sys.stderr)
        sys.exit(130)