feat(publish): P1-1 digest projection — global_digests/digest_topics → render-ready feed

/published/digest 가 read-time projection 반환: version 커서=global_digests.id
(일간 단일라이터 gapless 불요) · pub_id=digest:<date>(date-as-id) · tombstone 없음.
각 digest 에 digest_topics(rank/label/summary/country/article_count/importance) 조인.
엔벨로프 FeedResponse 재사용(뷰어 pull-sync 공용). DIGEST_PUBLISH_ENABLED 점등(host .env).
검증: since=70 → rev71/72 실데이터(49·54 토픽) · since=72 → 빈 배치 next_since 유지(증분 정확).
docsrv-viewer-publish 트랙 (plan viewer-daily-report P1-1).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/api/published.py

@@ -19,7 +19,7 @@ import logging
 
 from fastapi import APIRouter, Depends, Header, HTTPException, Query
 from pydantic import BaseModel
-from sqlalchemy import select
+from sqlalchemy import select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from core.config import settings
@@ -113,13 +113,13 @@ async def published_feed(
     )
 
 
-# ── P1-1: 뉴스/다이제스트 발행 read API (scaffold, docsrv-viewer-publish) ──────────
-# 첫 실사용 = 추상화 적합성 시험(plan r2). 다이제스트는 문서용 불변식 일부 제외(content-type
-# 파라미터화): rev=단순 version 커서(증분 pull) · pub_id=date-as-id(admin-gated feed 라 opacity
-# 불필요) · tombstone 없음(다이제스트 미삭제). 엔벨로프는 /feed 와 동일(FeedResponse) → 뷰어
-# pull-sync 클라이언트 재사용. scaffold-first: DIGEST_PUBLISH_ENABLED off(기본)=503(명시적 미가동,
-# no-silent). 점등돼도 projection 미구현이면 503 — 실데이터·시크릿 0.
-DIGEST_FEED_SCHEMA_VERSION = 1
+# ── P1-1: 뉴스/다이제스트 발행 read API (docsrv-viewer-publish) ────────────────────
+# global_digests(일간 컨테이너) + digest_topics(토픽 N, digest_id FK) -> render-ready
+# read-time projection. content-type 파라미터화(plan r2): version 커서=global_digests.id
+# (일간 단일 라이터라 gapless 불요·gap 무해) · pub_id=date-as-id(admin-gated feed 라 opacity
+# 불필요) · tombstone 없음(다이제스트 미삭제). 엔벨로프는 /feed 와 동일(FeedResponse)=뷰어 재사용.
+# scaffold-first: DIGEST_PUBLISH_ENABLED off(기본)=503(명시적 미가동, no-silent).
+DIGEST_PAYLOAD_SCHEMA_VERSION = 1
 
 
 @router.get("/digest", response_model=FeedResponse)
@@ -127,9 +127,72 @@ async def published_digest(
     since: int = Query(0, ge=0),
     limit: int = Query(DEFAULT_LIMIT, ge=1, le=MAX_LIMIT),
     _auth: None = Depends(_verify_token),
+    session: AsyncSession = Depends(_session),
 ):
-    """뉴스/다이제스트 발행 feed (version 커서 since). scaffold — 점등/projection 전 503."""
+    """global_digests.id > since 를 id ASC 로 limit 만큼. 각 digest 에 topics 조인해 render-ready 반환."""
     if not settings.digest_publish_enabled:
         raise HTTPException(status_code=503, detail="digest publish not enabled (scaffold)")
-    # TODO(P1 후속): global_digests/digest_topics -> render-ready projection (version 커서).
-    raise HTTPException(status_code=503, detail="digest publish projection not implemented (scaffold)")
+
+    drows = (await session.execute(
+        text(
+            "SELECT id, digest_date, status, total_articles, total_topics, total_countries, created_at "
+            "FROM global_digests WHERE id > :since ORDER BY id ASC LIMIT :limit"
+        ),
+        {"since": since, "limit": limit},
+    )).mappings().all()
+
+    if not drows:
+        return FeedResponse(schema_version=FEED_SCHEMA_VERSION, items=[], next_since=since, has_more=False)
+
+    ids = [r["id"] for r in drows]
+    trows = (await session.execute(
+        text(
+            "SELECT digest_id, topic_rank, topic_label, summary, country, article_count, importance_score "
+            "FROM digest_topics WHERE digest_id = ANY(:ids) ORDER BY digest_id ASC, topic_rank ASC"
+        ),
+        {"ids": ids},
+    )).mappings().all()
+
+    topics_by_digest: dict[int, list[dict]] = {}
+    for t in trows:
+        topics_by_digest.setdefault(t["digest_id"], []).append({
+            "rank": t["topic_rank"],
+            "label": t["topic_label"],
+            "summary": t["summary"],
+            "country": t["country"],
+            "article_count": t["article_count"],
+            "importance": t["importance_score"],
+        })
+
+    items = []
+    for r in drows:
+        d_date = r["digest_date"].isoformat() if r["digest_date"] else None
+        items.append(FeedItem(
+            pub_id=f"digest:{d_date}",
+            kind="digest",
+            source_id=r["id"],
+            rev=r["id"],
+            deleted=False,
+            schema_version=DIGEST_PAYLOAD_SCHEMA_VERSION,
+            payload={
+                "digest_date": d_date,
+                "status": r["status"],
+                "total_articles": r["total_articles"],
+                "total_topics": r["total_topics"],
+                "total_countries": r["total_countries"],
+                "generated_at": r["created_at"].isoformat() if r["created_at"] else None,
+                "topics": topics_by_digest.get(r["id"], []),
+            },
+        ))
+    next_since = items[-1].rev
+    has_more = len(drows) == limit
+    logger.info(
+        "published_digest since=%s returned=%s next_since=%s has_more=%s",
+        since, len(items), next_since, has_more,
+    )
+    return FeedResponse(
+        schema_version=FEED_SCHEMA_VERSION,
+        items=items,
+        next_since=next_since,
+        has_more=has_more,
+    )