feat(meals): dual-read week keys and meal-to-Flex join (PLAT-134) (#182)

* feat(meals): dual-read week keys and meal-to-Flex join (PLAT-134)

* style(meals): apply ruff format (PLAT-134)

* docs(meals): clarify week-key dual-read is same-instant (PLAT-134)

src/shared/shared/db.py

@@ -7,6 +7,8 @@ from zoneinfo import ZoneInfo
 from boto3.dynamodb.conditions import Attr, Key
 import boto3
 
+from .week_keys import candidate_week_ids
+
 EASTERN = ZoneInfo("America/New_York")
 _table = None
 
@@ -49,9 +51,27 @@ def put_menu(week: str, menu_data: dict):
     )
 
 
-def get_menu(week: str) -> dict | None:
-    resp = _get_table().get_item(Key={"PK": f"WEEK#{week}", "SK": "MENU"})
-    return resp.get("Item")
+def _week_ids_to_read(week: str | datetime) -> tuple[str, ...]:
+    """Resolve week ids to try. Datetime dual-reads U-week then ISO; a string is used as-is."""
+    if isinstance(week, datetime):
+        return candidate_week_ids(week)
+    return (week,)
+
+
+def _get_item_for_weeks(
+    week: str, sk: str, when: datetime | None = None
+) -> dict | None:
+    table = _get_table()
+    for week_id in _week_ids_to_read(when if when is not None else week):
+        resp = table.get_item(Key={"PK": f"WEEK#{week_id}", "SK": sk})
+        item = resp.get("Item")
+        if item:
+            return item
+    return None
+
+
+def get_menu(week: str, when: datetime | None = None) -> dict | None:
+    return _get_item_for_weeks(week, "MENU", when=when)
 
 
 def set_form_status(week: str, status: str):
@@ -62,8 +82,8 @@ def set_form_status(week: str, status: str):
     )
 
 
-def get_form_status(week: str) -> str:
-    menu = get_menu(week)
+def get_form_status(week: str, when: datetime | None = None) -> str:
+    menu = get_menu(week, when=when)
     if not menu:
         return "not_found"
     return menu.get("form_status", "closed")
@@ -91,19 +111,23 @@ def put_order(week: str, employee_slug: str, order_data: dict):
     )
 
 
-def get_orders(week: str) -> list[dict]:
-    resp = _get_table().query(
-        KeyConditionExpression=Key("PK").eq(f"WEEK#{week}")
+def get_orders(week: str, when: datetime | None = None) -> list[dict]:
+    table = _get_table()
+    for week_id in _week_ids_to_read(when if when is not None else week):
+        resp = table.query(
+            KeyConditionExpression=Key("PK").eq(f"WEEK#{week_id}")
             & Key("SK").begins_with("ORDER#"),
         )
-    return resp.get("Items", [])
+        items = resp.get("Items", [])
+        if items:
+            return items
+    return []
 
 
-def get_order(week: str, employee_slug: str) -> dict | None:
-    resp = _get_table().get_item(
-        Key={"PK": f"WEEK#{week}", "SK": f"ORDER#{employee_slug}"}
-    )
-    return resp.get("Item")
+def get_order(
+    week: str, employee_slug: str, when: datetime | None = None
+) -> dict | None:
+    return _get_item_for_weeks(week, f"ORDER#{employee_slug}", when=when)
 
 
 def delete_order(week: str, employee_slug: str):
@@ -123,9 +147,8 @@ def put_summary(week: str, summary: dict):
     )
 
 
-def get_summary(week: str) -> dict | None:
-    resp = _get_table().get_item(Key={"PK": f"WEEK#{week}", "SK": "SUMMARY"})
-    return resp.get("Item")
+def get_summary(week: str, when: datetime | None = None) -> dict | None:
+    return _get_item_for_weeks(week, "SUMMARY", when=when)
 
 
 def get_roster() -> list[dict]:

src/shared/shared/flex_join.py

@@ -0,0 +1,110 @@
+"""Dry-run join of meal orders to Flex worker emails.
+
+Join meal ``employee_email`` to Flex worker emails: BUSINESS first, then
+personal. ``matched_on="business"`` wins if both would match.
+
+A live Flex worker pull is a later dry-run the operator runs. This module
+accepts a worker-email fixture and does not call Paychex or POST
+checkcomponents. It does not map WEEK# onto Flex payPeriodId.
+
+Thursday close dates are not meal weeks for a Flex pay period; that mapping
+is not encoded here.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Literal
+
+MatchedOn = Literal["business", "personal"]
+
+
+@dataclass(frozen=True)
+class JoinRow:
+    employee_email: str
+    worker_id: str
+    matched_on: MatchedOn
+
+
+@dataclass(frozen=True)
+class JoinReport:
+    matched: tuple[JoinRow, ...]
+    unmatched: tuple[str, ...]
+    week: str | None = None
+
+
+def _normalize_email(value: object) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip().lower()
+    return text or None
+
+
+def _emails_from_worker(worker: dict[str, Any]) -> tuple[str | None, str | None]:
+    business = _normalize_email(worker.get("business_email"))
+    personal = _normalize_email(worker.get("personal_email"))
+    communications = worker.get("communications") or []
+    if isinstance(communications, list):
+        for comm in communications:
+            if not isinstance(comm, dict):
+                continue
+            ctype = str(comm.get("type") or "").strip().upper()
+            email = _normalize_email(comm.get("uri"))
+            if not email:
+                continue
+            if ctype == "BUSINESS" and business is None:
+                business = email
+            elif ctype in {"PERSONAL", "HOME"} and personal is None:
+                personal = email
+    return business, personal
+
+
+def join_orders_to_workers(
+    orders: list[dict[str, Any]],
+    workers: list[dict[str, Any]],
+    *,
+    week: str | None = None,
+) -> JoinReport:
+    business_index: dict[str, str] = {}
+    personal_index: dict[str, str] = {}
+    for worker in workers:
+        worker_id = worker.get("workerId")
+        if worker_id is None:
+            continue
+        worker_id_str = str(worker_id)
+        business, personal = _emails_from_worker(worker)
+        if business is not None and business not in business_index:
+            business_index[business] = worker_id_str
+        if personal is not None and personal not in personal_index:
+            personal_index[personal] = worker_id_str
+
+    matched: list[JoinRow] = []
+    unmatched: list[str] = []
+    for order in orders:
+        raw_email = order.get("employee_email")
+        original = "" if raw_email is None else str(raw_email)
+        key = _normalize_email(raw_email)
+        if key is not None and key in business_index:
+            matched.append(
+                JoinRow(
+                    employee_email=original,
+                    worker_id=business_index[key],
+                    matched_on="business",
+                )
+            )
+        elif key is not None and key in personal_index:
+            matched.append(
+                JoinRow(
+                    employee_email=original,
+                    worker_id=personal_index[key],
+                    matched_on="personal",
+                )
+            )
+        else:
+            unmatched.append(original)
+
+    return JoinReport(
+        matched=tuple(matched),
+        unmatched=tuple(unmatched),
+        week=week,
+    )

src/shared/shared/week_keys.py

@@ -0,0 +1,41 @@
+"""Sunday-week and ISO week id helpers for Dynamo WEEK# dual-read.
+
+Existing meal PKs were written with strftime("%Y-W%U") (Sunday week, can be
+W00). ISO week is "%G-W%V". Dual-read tries U-week first, then ISO, without a
+migration. Writes stay on %Y-W%U.
+
+This is not a mapping from WEEK# onto Flex payPeriodId.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime
+
+WEEK_ID_RE = re.compile(r"^\d{4}-W\d{2}$")
+
+
+def week_id_ustrftime(dt: datetime) -> str:
+    return dt.strftime("%Y-W%U")
+
+
+def week_id_iso(dt: datetime) -> str:
+    return dt.strftime("%G-W%V")
+
+
+def candidate_week_ids(dt: datetime) -> tuple[str, ...]:
+    """Return U-week then ISO encodings of the same instant.
+
+    Dual-read does not advance the calendar week. ``%Y-W%U`` and ``%G-W%V``
+    are two strftime renderings of ``dt``. Example: 2026-01-01 encodes as
+    ("2026-W00", "2026-W01"), not as "this week" then "next week".
+    """
+    u_week = week_id_ustrftime(dt)
+    iso_week = week_id_iso(dt)
+    if u_week == iso_week:
+        return (u_week,)
+    return (u_week, iso_week)
+
+
+def candidate_week_pks(dt: datetime) -> tuple[str, ...]:
+    return tuple(f"WEEK#{week_id}" for week_id in candidate_week_ids(dt))

tests/test_flex_join.py

@@ -0,0 +1,161 @@
+"""Tests for meal-order to Flex worker email join (fixture-driven, no HTTP)."""
+
+import inspect
+
+from shared.flex_join import JoinReport, JoinRow, join_orders_to_workers
+
+
+def test_business_email_match():
+    orders = [
+        {"employee_email": "pat@seahaven.com", "employee_name": "Pat", "total": 12}
+    ]
+    workers = [
+        {
+            "workerId": "W1",
+            "business_email": "pat@seahaven.com",
+            "personal_email": "pat@gmail.com",
+        }
+    ]
+
+    report = join_orders_to_workers(orders, workers, week="2026-W34")
+
+    assert report == JoinReport(
+        matched=(
+            JoinRow(
+                employee_email="pat@seahaven.com",
+                worker_id="W1",
+                matched_on="business",
+            ),
+        ),
+        unmatched=(),
+        week="2026-W34",
+    )
+
+
+def test_fallback_to_personal_when_business_missing():
+    orders = [{"employee_email": "pat@gmail.com"}]
+    workers = [
+        {
+            "workerId": "W1",
+            "business_email": "",
+            "personal_email": "  PAT@gmail.com ",
+        }
+    ]
+
+    report = join_orders_to_workers(orders, workers)
+
+    assert report.matched == (
+        JoinRow(
+            employee_email="pat@gmail.com",
+            worker_id="W1",
+            matched_on="personal",
+        ),
+    )
+    assert report.unmatched == ()
+
+
+def test_fallback_to_personal_when_business_mismatches():
+    orders = [{"employee_email": "alex@gmail.com"}]
+    workers = [
+        {
+            "workerId": "W2",
+            "business_email": "alex@seahaven.com",
+            "personal_email": "alex@gmail.com",
+        }
+    ]
+
+    report = join_orders_to_workers(orders, workers)
+
+    assert report.matched[0].matched_on == "personal"
+    assert report.matched[0].worker_id == "W2"
+    assert report.unmatched == ()
+
+
+def test_unmatched_leftover():
+    orders = [
+        {"employee_email": "pat@seahaven.com"},
+        {"employee_email": "unknown@example.com"},
+    ]
+    workers = [
+        {
+            "workerId": "W1",
+            "business_email": "pat@seahaven.com",
+            "personal_email": "pat@gmail.com",
+        }
+    ]
+
+    report = join_orders_to_workers(orders, workers)
+
+    assert len(report.matched) == 1
+    assert report.unmatched == ("unknown@example.com",)
+
+
+def test_business_wins_over_personal_if_both_present():
+    orders = [{"employee_email": "sam@seahaven.com"}]
+    workers = [
+        {
+            "workerId": "W-personal",
+            "business_email": "other@seahaven.com",
+            "personal_email": "sam@seahaven.com",
+        },
+        {
+            "workerId": "W-business",
+            "business_email": "sam@seahaven.com",
+            "personal_email": "sam@gmail.com",
+        },
+    ]
+
+    report = join_orders_to_workers(orders, workers)
+
+    assert report.matched == (
+        JoinRow(
+            employee_email="sam@seahaven.com",
+            worker_id="W-business",
+            matched_on="business",
+        ),
+    )
+    assert report.unmatched == ()
+
+
+def test_same_worker_business_and_personal_same_email_is_business():
+    orders = [{"employee_email": "lee@seahaven.com"}]
+    workers = [
+        {
+            "workerId": "W3",
+            "business_email": "lee@seahaven.com",
+            "personal_email": "lee@seahaven.com",
+        }
+    ]
+
+    report = join_orders_to_workers(orders, workers)
+
+    assert report.matched[0].matched_on == "business"
+    assert report.matched[0].worker_id == "W3"
+
+
+def test_nested_communications_used_when_flat_keys_missing():
+    orders = [{"employee_email": "pat@seahaven.com"}]
+    workers = [
+        {
+            "workerId": "W1",
+            "communications": [
+                {"type": "BUSINESS", "uri": "pat@seahaven.com"},
+                {"type": "PERSONAL", "uri": "pat@gmail.com"},
+            ],
+        }
+    ]
+
+    report = join_orders_to_workers(orders, workers)
+
+    assert report.matched[0].matched_on == "business"
+    assert report.matched[0].worker_id == "W1"
+
+
+def test_flex_join_has_no_paychex_http():
+    import shared.flex_join as flex_join
+
+    source = inspect.getsource(flex_join)
+    assert "urllib" not in source
+    assert "api.paychex.com" not in source
+    assert "requests" not in source
+    assert "http.client" not in source

tests/test_week_keys.py

@@ -0,0 +1,102 @@
+"""Tests for Sunday-week / ISO dual-read week keys and GET helpers."""
+
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+from shared import db
+from shared.week_keys import (
+    WEEK_ID_RE,
+    candidate_week_ids,
+    candidate_week_pks,
+    week_id_iso,
+    week_id_ustrftime,
+)
+
+MISMATCH_DT = datetime(2026, 1, 1)
+MATCH_DT = datetime(2026, 1, 4)
+
+
+def test_ustrftime_and_iso_differ_on_new_year():
+    assert week_id_ustrftime(MISMATCH_DT) == "2026-W00"
+    assert week_id_iso(MISMATCH_DT) == "2026-W01"
+    assert WEEK_ID_RE.match("2026-W00")
+    assert WEEK_ID_RE.match("2026-W01")
+
+
+def test_candidate_week_ids_u_week_first_then_iso():
+    # 2026-01-01: same instant encoded two ways, not a next-week lookup.
+    assert candidate_week_ids(MISMATCH_DT) == ("2026-W00", "2026-W01")
+
+
+def test_candidate_week_ids_unique_when_formats_match():
+    ids = candidate_week_ids(MATCH_DT)
+    assert week_id_ustrftime(MATCH_DT) == week_id_iso(MATCH_DT) == "2026-W01"
+    assert ids == ("2026-W01",)
+    assert len(ids) == 1
+
+
+def test_candidate_week_pks_prefixed():
+    assert candidate_week_pks(MISMATCH_DT) == ("WEEK#2026-W00", "WEEK#2026-W01")
+    assert candidate_week_pks(MATCH_DT) == ("WEEK#2026-W01",)
+
+
+def _table_with_get_items(items_by_pk: dict[str, dict]):
+    table = MagicMock()
+
+    def get_item(Key):
+        item = items_by_pk.get(Key["PK"])
+        if item is None:
+            return {}
+        if item.get("SK") != Key["SK"]:
+            return {}
+        return {"Item": item}
+
+    table.get_item.side_effect = get_item
+    return table
+
+
+def test_get_menu_when_falls_back_to_iso():
+    iso_item = {"PK": "WEEK#2026-W01", "SK": "MENU", "source": "iso"}
+    table = _table_with_get_items({"WEEK#2026-W01": iso_item})
+
+    with patch.object(db, "_get_table", return_value=table):
+        result = db.get_menu("2026-W00", when=MISMATCH_DT)
+
+    assert result == iso_item
+    pks = [call.kwargs["Key"]["PK"] for call in table.get_item.call_args_list]
+    assert pks == ["WEEK#2026-W00", "WEEK#2026-W01"]
+
+
+def test_get_menu_when_prefers_u_week_if_both_exist():
+    u_item = {"PK": "WEEK#2026-W00", "SK": "MENU", "source": "u"}
+    iso_item = {"PK": "WEEK#2026-W01", "SK": "MENU", "source": "iso"}
+    table = _table_with_get_items({"WEEK#2026-W00": u_item, "WEEK#2026-W01": iso_item})
+
+    with patch.object(db, "_get_table", return_value=table):
+        result = db.get_menu("2026-W00", when=MISMATCH_DT)
+
+    assert result == u_item
+    assert table.get_item.call_count == 1
+    assert table.get_item.call_args.kwargs["Key"]["PK"] == "WEEK#2026-W00"
+
+
+def test_get_menu_without_when_uses_week_argument_as_single_key():
+    table = _table_with_get_items({})
+
+    with patch.object(db, "_get_table", return_value=table):
+        result = db.get_menu("2026-W34")
+
+    assert result is None
+    table.get_item.assert_called_once_with(Key={"PK": "WEEK#2026-W34", "SK": "MENU"})
+
+
+def test_get_orders_when_falls_back_to_iso():
+    iso_orders = [{"PK": "WEEK#2026-W01", "SK": "ORDER#pat"}]
+    table = MagicMock()
+    table.query.side_effect = [{"Items": []}, {"Items": iso_orders}]
+
+    with patch.object(db, "_get_table", return_value=table):
+        result = db.get_orders("2026-W00", when=MISMATCH_DT)
+
+    assert result == iso_orders
+    assert table.query.call_count == 2