/.github/scripts/rotation.py

#!/usr/bin/env python3
"""Daily Discord-watch rotation reminder.

Picks the person on watch for the current day and pings them in Slack on the
morning of *their* local timezone. The rotation is deterministic — the
assignee is a function of the date and the person's position in the list — so
there is no state to store anywhere.

The GitHub Actions workflow wakes at a couple of fixed UTC times (one per
timezone's morning). On each run the day's assignee is pinged only if it's
currently morning where they live; if not, the run for their timezone's
morning handles them. Our timezones are far enough apart that only one is ever
in its morning at a time, so at most one person is pinged per run.

Set SLACK_WEBHOOK_URL to post for real. Leave it unset for a dry run that just
prints what it would do — handy for testing the rotation order without Slack.
"""

from __future__ import annotations

import datetime
import json
import os
import urllib.error
import urllib.request
from dataclasses import dataclass
from zoneinfo import ZoneInfo

# Each cron run is one timezone's morning scan: we ping today's assignee only
# if it's currently morning where they are. A run that's morning in SF is night
# in Singapore and vice versa, so at most one timezone matches per run. Morning
# is a band rather than an exact hour, which absorbs both daylight saving and
# GitHub's frequently-delayed cron schedule — a run that fires a few hours late
# still counts as that person's morning. The band starts at 05:00 (not
# midnight) so a delayed *other* timezone's cron spilling past local midnight
# isn't mistaken for this timezone's morning, which would double-ping.
MORNING_START_HOUR = 5
MORNING_END_HOUR = 12

# Skip Saturdays and Sundays (in each person's local time). The rotation also
# advances by workdays only, so Friday hands off straight to Monday.
WEEKDAYS_ONLY = True

# Rotation anchor: workday 0 is this date. Any Monday works; it only sets the
# phase of the cycle, not who is in it.
EPOCH = datetime.date(2026, 1, 5)  # a Monday


@dataclass(frozen=True)
class Person:
    name: str  # for logs / dry-run output only
    slack_id: str  # Slack member ID, e.g. "U01ABC2DEF" (NOT the display name)
    tz: str  # IANA timezone name, e.g. "America/Los_Angeles"
    # Out-of-office spans as inclusive (start, end) ISO date pairs, e.g.
    # (("2026-07-13", "2026-07-17"),). On any OOO day the person is skipped and
    # the next available person covers; the OOO person keeps their later slots.
    ooo: tuple[tuple[str, str], ...] = ()


# Rotation order. Slack member IDs (profile -> ⋮ More -> Copy member ID) and
# each person's IANA timezone.
PEOPLE: list[Person] = [
    Person("Aravind Segu", "U01A12R8NUR", "America/Los_Angeles"),
    Person("Bryan Qiu", "U05KA5T983Y", "America/Los_Angeles"),
    Person("Daniel Lok", "U060CNWNHSQ", "Asia/Singapore"),
    Person("Dhruv Gupta", "U0A76097E1F", "America/Los_Angeles"),
    Person("Edwin He", "U077B1V6WQJ", "America/Los_Angeles"),
    Person("Pat Sukprasert", "U05HRKWFY81", "Asia/Singapore"),
    Person("Sabhya Chhabria", "U07A1KQDXAB", "America/Los_Angeles"),
    Person("Serena Ruan", "U0571L5KNLR", "Asia/Singapore"),
    Person("Shivam Mittal", "U09FZKX9S6B", "America/Los_Angeles"),
    Person("Tomu Hirata", "U07TX4PR5MZ", "Asia/Singapore"),
    Person("Zeyi (Rice) Fan", "U09L5HT4CH0", "America/Los_Angeles"),
]


def _workdays_between(start: datetime.date, end: datetime.date) -> int:
    """Number of Mon–Fri days in [start, end). Negative if end precedes start."""
    if end < start:
        return -_workdays_between(end, start)
    full_weeks, extra = divmod((end - start).days, 7)
    count = full_weeks * 5
    for i in range(extra):
        if (start + datetime.timedelta(days=full_weeks * 7 + i)).weekday() < 5:
            count += 1
    return count


def is_ooo(person: Person, local_date: datetime.date) -> bool:
    """Whether person is out of office on local_date (inclusive spans)."""
    for start, end in person.ooo:
        if datetime.date.fromisoformat(start) <= local_date <= datetime.date.fromisoformat(end):
            return True
    return False


def assignee_for(local_date: datetime.date) -> Person | None:
    """The person on watch for a given local workday, or None if all are OOO.

    Indexed by the number of workdays since EPOCH (which is itself a Monday),
    so weekends advance nobody and Friday hands off directly to Monday. If the
    slot's person is OOO, the next available person covers — probing forward so
    coverage stays a pure function of the date (no stored state). Only
    meaningful for weekdays; weekends are filtered out before this is called.
    """
    workday_number = _workdays_between(EPOCH, local_date)
    for offset in range(len(PEOPLE)):
        person = PEOPLE[(workday_number + offset) % len(PEOPLE)]
        if not is_ooo(person, local_date):
            return person
    return None  # everyone is OOO that day


def whose_turn_now(now_utc: datetime.datetime) -> Person | None:
    """Return the person to ping right now, or None if it isn't anyone's morning.

    Each person is evaluated in their own timezone: it must be a weekday morning
    (before noon) there, and today's rotation slot must land on them. Since our
    timezones are far enough apart that only one is ever in its morning at a
    time, at most one person matches. A person missed by a late/early run is
    picked up by the next run that lands in their morning.
    """
    for person in PEOPLE:
        local = now_utc.astimezone(ZoneInfo(person.tz))
        if not (MORNING_START_HOUR <= local.hour < MORNING_END_HOUR):
            continue
        if WEEKDAYS_ONLY and local.weekday() >= 5:  # 5=Sat, 6=Sun
            continue
        if assignee_for(local.date()) == person:
            return person
    return None


class SlackPostError(RuntimeError):
    """Raised when the Slack POST fails, without exposing the webhook URL."""


def post_to_slack(webhook_url: str, person: Person) -> None:
    text = (
        f"<@{person.slack_id}> you're on *Discord watch* today \U0001f440 "
        f"— please keep an eye on the channel."
    )
    payload = json.dumps({"text": text}).encode()
    req = urllib.request.Request(
        webhook_url,
        data=payload,
        headers={"Content-Type": "application/json"},
    )
    # Catch and re-raise without the URL: urllib errors stringify the full
    # webhook URL, which must never reach the Actions log or error output.
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            resp.read()
    except urllib.error.HTTPError as exc:
        raise SlackPostError(f"Slack returned HTTP {exc.code} {exc.reason}") from None
    except urllib.error.URLError as exc:
        raise SlackPostError(f"could not reach Slack: {exc.reason}") from None


def _report_todays_assignees(now_utc: datetime.datetime) -> None:
    """Log who's on watch for each timezone's current local date.

    Runs regardless of the morning window so a manual run is always
    informative, even outside anyone's ping window.
    """
    for tz in sorted({p.tz for p in PEOPLE}):
        local = now_utc.astimezone(ZoneInfo(tz))
        if local.weekday() >= 5:  # 5=Sat, 6=Sun
            who = "nobody (weekend)"
        else:
            person = assignee_for(local.date())
            who = person.name if person else "nobody (all OOO)"
        print(f"  {tz}: {local:%Y-%m-%d %a} -> {who}")


def main() -> None:
    now_utc = datetime.datetime.now(datetime.timezone.utc)

    print(f"Today's watch by timezone (as of {now_utc:%Y-%m-%d %H:%M UTC}):")
    _report_todays_assignees(now_utc)

    person = whose_turn_now(now_utc)

    if person is None:
        print(f"{now_utc:%Y-%m-%d %H:%M UTC}: nobody's on watch right now, nothing to do.")
        return

    local = now_utc.astimezone(ZoneInfo(person.tz))
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    if not webhook_url:
        print(
            f"[dry run] Would ping {person.name} ({person.slack_id}) "
            f"— it's {local:%Y-%m-%d %H:%M} in {person.tz}. "
            f"Set SLACK_WEBHOOK_URL to post for real."
        )
        return

    post_to_slack(webhook_url, person)
    print(f"Pinged {person.name} ({person.slack_id}) at {local:%Y-%m-%d %H:%M %Z}.")


if __name__ == "__main__":
    main()