[카테고리:] iOS — Push / Live Activity

전체 구조

iEvent의 푸시/LiveActivity 시스템은 iOS 앱, FastAPI 백엔드, Python Scheduler 세 개의 독립 프로세스로 구성됩니다.

iOS App (ActivityKit)
    ↕  APNs (Apple Push Notification service)
       ↑
FastAPI Backend  ←→  MySQL / Redis
       ↑
Python Scheduler (독립 프로세스, 매 분 실행)

Scheduler를 독립 프로세스로 분리한 이유

• Event loop 보호: “종료 15분 전” 크론 작업을 FastAPI 내부에서 돌리면 event loop 블로킹 위험
• 독립 배포: Scheduler 버그로 재시작해도 API 서버는 무중단 유지
• 리소스 격리: LLM 호출처럼 CPU를 많이 쓰는 작업을 API 서버와 분리

iOS 측: LiveActivity 등록

// LiveActivityManager.swift
struct IVENTActivityAttributes: ActivityAttributes {
    struct ContentState: Codable, Hashable {
        var childName: String
        var scheduleTitle: String
        var endTime: Date
        var status: String
    }
    var childId: String
    var academyName: String
}

func startActivity(for schedule: Schedule) async throws {
    guard ActivityAuthorizationInfo().areActivitiesEnabled else { return }

    let attributes = IVENTActivityAttributes(
        childId: schedule.childId,
        academyName: schedule.academyName
    )
    let state = IVENTActivityAttributes.ContentState(
        childName: schedule.childName,
        scheduleTitle: schedule.title,
        endTime: schedule.endTime,
        status: "active"
    )
    let activity = try Activity.request(
        attributes: attributes,
        content: .init(state: state, staleDate: schedule.endTime.addingTimeInterval(300))
    )
    if let token = activity.pushToken {
        await registerPushToken(token, activityId: activity.id)
    }
}

백엔드: ES256 JWT로 APNs 인증

# apns_client.py
def _make_jwt() -> str:
    from jose import jwt as jose_jwt
    key_path = settings.APNS_AUTH_KEY_PATH   # K8s Secret으로 주입된 .p8 파일
    with open(key_path, "r") as f:
        private_key = f.read()
    headers = {"kid": settings.APNS_KEY_ID}
    payload = {"iss": settings.APNS_TEAM_ID, "iat": int(time.time())}
    return jose_jwt.encode(payload, private_key, algorithm="ES256", headers=headers)

async def send_live_activity_update(self, push_token, event, content_state, environment=None):
    host = self._host_for(environment)
    topic = f"{self._bundle_id}.push-type.liveactivity"
    payload = {
        "aps": {
            "timestamp": int(time.time()),
            "event": event,   # "update" | "end"
            "content-state": content_state,
        }
    }
    headers = {
        "authorization": f"bearer {_make_jwt()}",
        "apns-push-type": "liveactivity",
        "apns-topic": topic,
        "apns-priority": "10",
    }
    async with httpx.AsyncClient(http2=True) as client:
        resp = await client.post(f"{host}/3/device/{push_token}",
                                 headers=headers, content=json.dumps(payload))
    return {"status_code": resp.status_code, "body": resp.text}

Scheduler: 매 분 체크

# push_notification_job.py
class PushNotificationJob(BaseJob):
    cron = "* * * * *"

    async def run(self, ctx: JobContext) -> JobResult:
        now_kst = datetime.now(KST)
        ending_soon = db.query(Schedule).filter(
            Schedule.end_time.between(now_kst, now_kst + timedelta(minutes=16)),
            Schedule.live_activity_token.isnot(None),
            Schedule.status == "active"
        ).all()

        for schedule in ending_soon:
            await apns_client.send_live_activity_update(
                push_token=schedule.live_activity_token,
                event="update",
                content_state={
                    "childName": schedule.child_name,
                    "scheduleTitle": schedule.title,
                    "endTime": schedule.end_time.timestamp(),
                    "status": "ending"
                },
                environment=schedule.apns_environment
            )
        return JobResult(success=True, processed=len(ending_soon))

교훈

APNs는 HTTP/2 프로토콜을 요구합니다. Python에서는 httpx에 http2=True를 반드시 지정해야 합니다. JWT는 생성 후 1시간 내에 사용해야 하므로 매 요청마다 새로 생성하는 것이 가장 안전합니다.