Skipping Tiles With No New Source Data

-- What produced each tile, and from what. One row per tile per build.
CREATE TABLE tile_builds (
    z            smallint    NOT NULL,
    x            integer     NOT NULL,
    y            integer     NOT NULL,
    geom         geometry(Polygon, 3857) NOT NULL,   -- the tile's own footprint
    source_state jsonb       NOT NULL,   -- {"scene_id": "digest", …} at build time
    built_at     timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (z, x, y)
);

CREATE INDEX tile_builds_gist ON tile_builds USING gist (geom);

-- Current source inventory: footprint and content digest per scene.
CREATE TABLE source_scenes (
    scene_id   text PRIMARY KEY,
    geom       geometry(Polygon, 3857) NOT NULL,
    digest     text        NOT NULL,
    updated_at timestamptz NOT NULL
);

CREATE INDEX source_scenes_gist ON source_scenes USING gist (geom);

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterator

import psycopg

# Metres of reach at this zoom, from the resampling kernel. A tile whose edge sits
# inside a changed scene's buffer changes even if its centre does not.
KERNEL_REACH_PX = {"nearest": 0, "bilinear": 1, "cubic": 2, "lanczos": 3}


STALE_TILES = """
WITH buffered AS (
    SELECT scene_id, digest, ST_Buffer(geom, %(reach_m)s) AS geom
    FROM   source_scenes
),
touching AS (
    SELECT t.z, t.x, t.y, t.source_state,
           jsonb_object_agg(b.scene_id, b.digest) AS current_state
    FROM   tile_builds t
    JOIN   buffered b ON ST_Intersects(t.geom, b.geom)
    WHERE  t.z = %(zoom)s
    GROUP  BY t.z, t.x, t.y, t.source_state
)
SELECT z, x, y, source_state, current_state
FROM   touching
-- The whole predicate: the set of (scene, digest) pairs touching this tile now
-- differs from the set that produced it. Comparing objects, not timestamps.
WHERE  source_state IS DISTINCT FROM current_state
"""


@dataclass(frozen=True)
class TileDecision:
    z: int
    x: int
    y: int
    rebuild: bool
    reason: str


def tiles_to_rebuild(conn: psycopg.Connection, zoom: int,
                     resampling: str = "bilinear") -> Iterator[TileDecision]:
    reach_m = KERNEL_REACH_PX.get(resampling, 1) * resolution_at(zoom)
    with conn.cursor() as cur:
        cur.execute(STALE_TILES, {"zoom": zoom, "reach_m": reach_m})
        for z, x, y, was, now in cur:
            changed = sorted(
                scene for scene in set(was) | set(now)
                if was.get(scene) != now.get(scene)
            )
            yield TileDecision(z, x, y, True,
                               f"changed sources: {', '.join(changed[:5])}")


def record_build(conn: psycopg.Connection, z: int, x: int, y: int,
                 source_state: dict[str, str]) -> None:
    """Write the state the tile was built FROM, at the moment it was built."""
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO tile_builds (z, x, y, geom, source_state, built_at)
            VALUES (%s, %s, %s, ST_TileEnvelope(%s, %s, %s), %s::jsonb, now())
            ON CONFLICT (z, x, y) DO UPDATE
              SET source_state = EXCLUDED.source_state, built_at = now()
            """,
            (z, x, y, z, x, y, json.dumps(source_state, sort_keys=True)),
        )

def test_unchanged_world_skips_everything(conn, seeded_pyramid) -> None:
    assert list(tiles_to_rebuild(conn, zoom=12)) == []


def test_one_changed_scene_rebuilds_its_neighbourhood(conn, seeded_pyramid) -> None:
    update_scene_digest(conn, scene_id="S2_20260807_T33WXP", digest="sha256:new")
    decisions = list(tiles_to_rebuild(conn, zoom=12))

    ids = {(d.x, d.y) for d in decisions}
    assert ids >= seeded_pyramid.tiles_covering("S2_20260807_T33WXP")
    # …and strictly more than the footprint, because of the buffer.
    assert ids > seeded_pyramid.tiles_covering("S2_20260807_T33WXP")
    assert all("S2_20260807_T33WXP" in d.reason for d in decisions)


def test_a_tile_with_no_history_always_rebuilds(conn) -> None:
    """IS DISTINCT FROM must treat a missing record as different, not as equal."""
    insert_tile_without_state(conn, z=12, x=2100, y=1300)
    assert (2100, 1300) in {(d.x, d.y) for d in tiles_to_rebuild(conn, zoom=12)}

SELECT count(*) FILTER (WHERE built_at > now() - interval '1 day') AS rebuilt,
       count(*)                                                    AS total,
       round(100.0 * count(*) FILTER (WHERE built_at > now() - interval '1 day')
             / count(*), 1)                                        AS pct
FROM   tile_builds
WHERE  z = 12;