Add privacy-safe product analytics

Establish an idempotent analytics pipeline and internal decision dashboard while keeping event metadata allowlisted and account deletion enforceable.

src/main/kotlin/com/osglab/account/Application.kt

@@ -23,7 +23,10 @@ import com.osglab.account.features.admin.services.AdminBootstrapService
 import com.osglab.account.features.admin.services.AdminOperatorService
 import com.osglab.account.features.admin.services.AdminSessionService
 import com.osglab.account.features.admin.stats.repositories.AdminStatsRepository
+import com.osglab.account.features.admin.stats.repositories.AdminProductAnalyticsRepository
 import com.osglab.account.features.admin.stats.repositories.ExposedAdminStatsRepository
+import com.osglab.account.features.admin.stats.repositories.ExposedAdminProductAnalyticsRepository
+import com.osglab.account.features.admin.stats.services.AdminProductAnalyticsService
 import com.osglab.account.features.admin.stats.services.AdminStatsService
 import com.osglab.account.features.admin.users.repositories.AdminUsersRepository
 import com.osglab.account.features.admin.users.repositories.ExposedAdminUsersRepository
@@ -36,6 +39,12 @@ import com.osglab.account.features.account.AppleAccountReauthenticator
 import com.osglab.account.features.account.AppleRevocationOutboxProcessor
 import com.osglab.account.features.account.ExposedAccountRepository
 import com.osglab.account.features.account.accountRoutes
+import com.osglab.account.features.analytics.repositories.AnalyticsRepository
+import com.osglab.account.features.analytics.repositories.ExposedAnalyticsRepository
+import com.osglab.account.features.analytics.routes.analyticsRoutes
+import com.osglab.account.features.analytics.services.AnalyticsMaintenanceService
+import com.osglab.account.features.analytics.services.AnalyticsService
+import com.osglab.account.features.analytics.services.DefaultAnalyticsService
 import com.osglab.account.features.appleevents.AppleEventService
 import com.osglab.account.features.appleevents.AppleEventRepository
 import com.osglab.account.features.appleevents.AppleEventVerifier
@@ -102,6 +111,7 @@ import com.osglab.account.features.integrity.UnavailableAppleDeviceCheckClient
 import com.osglab.account.features.integrity.createDeviceCheckClient
 import com.osglab.account.features.integrity.integrityRoutes
 import com.osglab.account.features.inviteweb.InviteWebConfig
+import com.osglab.account.features.inviteweb.InviteOpenRecorder
 import com.osglab.account.features.inviteweb.ReferralLookupPort
 import com.osglab.account.features.inviteweb.configureInviteWebRoutes
 import com.osglab.account.features.referrals.domain.ReferralException
@@ -261,6 +271,13 @@ fun Application.module() {
             } catch (_: Exception) {
                 // Durable outbox state is retried; never log sensitive token material.
             }
+            try {
+                koin.get<AnalyticsMaintenanceService>().purgeStaleAnonymousInstallations()
+            } catch (exception: CancellationException) {
+                throw exception
+            } catch (_: Exception) {
+                // Anonymous analytics cleanup is bounded and retried on the next cycle.
+            }
             try {
                 koin.get<GatewayReconciliationService>().reconcile()
             } catch (exception: CancellationException) {
@@ -308,7 +325,8 @@ fun Application.module() {
         }
         rateLimit(PUBLIC_RATE_LIMIT) {
             appleEventRoutes(koin.get())
-            configureInviteWebRoutes(koin.get(), koin.get())
+            analyticsRoutes(koin.get())
+            configureInviteWebRoutes(koin.get(), koin.get(), koin.get())
             integrityRoutes(koin.get())
         }
         if (appConfig.admin.enabled) {
@@ -319,6 +337,7 @@ fun Application.module() {
                     authService = koin.get(),
                     sessionService = koin.get(),
                     statsService = koin.get(),
+                    productAnalyticsService = koin.get(),
                     usersService = koin.get(),
                     grantService = koin.get(),
                     operatorService = koin.get(),
@@ -393,6 +412,8 @@ fun accountServerModule(config: AppConfig): Module = module {
     single { AdminAuditService(get()) }
     single<AdminStatsRepository> { ExposedAdminStatsRepository(get()) }
     single { AdminStatsService(get()) }
+    single<AdminProductAnalyticsRepository> { ExposedAdminProductAnalyticsRepository(get()) }
+    single { AdminProductAnalyticsService(get()) }
     single<AdminUsersRepository> { ExposedAdminUsersRepository(get()) }
     single { AdminUsersService(get()) }
     single { AdminGrantService(get()) }
@@ -422,6 +443,14 @@ fun accountServerModule(config: AppConfig): Module = module {
     single<AuthRepository> { ExposedAuthRepository(get()) }
     single { SessionAccessAuthenticator(get(), get()) }
     single<AccountRepository> { ExposedAccountRepository(get(), get()) }
+    single<AnalyticsRepository> { ExposedAnalyticsRepository(get()) }
+    single<AnalyticsService> { DefaultAnalyticsService(get()) }
+    single { AnalyticsMaintenanceService(get()) }
+    single<InviteOpenRecorder> {
+        InviteOpenRecorder {
+            get<AnalyticsRepository>().recordInvitePageOpen(Instant.now())
+        }
+    }
     single<AppleEventRepository> { ExposedAppleEventRepository(get(), get(), config.antiAbuse) }
     single { AppleEventVerifier(config.apple, get()) }
     single { AppleEventService(get(), get()) }

src/main/kotlin/com/osglab/account/features/admin/routes/AdminRoutes.kt

@@ -18,6 +18,7 @@ import com.osglab.account.features.admin.services.AdminOperatorService
 import com.osglab.account.features.admin.services.AdminSessionService
 import com.osglab.account.features.admin.stats.models.AdminStatsDto
 import com.osglab.account.features.admin.stats.models.AdminUsageAggregateDto
+import com.osglab.account.features.admin.stats.services.AdminProductAnalyticsService
 import com.osglab.account.features.admin.stats.services.AdminStatsService
 import com.osglab.account.features.admin.users.models.AdminUserDetailDto
 import com.osglab.account.features.admin.users.models.AdminUserLedgerEntryDto
@@ -59,6 +60,7 @@ fun Route.adminApiRoutes(
     authService: AdminAuthService,
     sessionService: AdminSessionService,
     statsService: AdminStatsService,
+    productAnalyticsService: AdminProductAnalyticsService,
     usersService: AdminUsersService,
     grantService: AdminGrantService,
     operatorService: AdminOperatorService,
@@ -158,6 +160,16 @@ fun Route.adminApiRoutes(
             call.respond(stats.toReferralResponse())
         }
 
+        get("/analytics") {
+            if (call.requirePrincipal(sessionService) == null) return@get
+            val window = parseAdminStatsRange(call.request.queryParameters["range"], clock)
+                ?: run {
+                    call.respond(HttpStatusCode.BadRequest, AdminErrorResponse("VALIDATION_ERROR"))
+                    return@get
+                }
+            call.respond(productAnalyticsService.get(window.first, window.second))
+        }
+
         get("/users") {
             if (
                 call.requireRole(
@@ -500,6 +512,11 @@ private suspend fun AdminStatsService.getRange(
     range: String?,
     clock: Clock,
 ): AdminStatsDto? {
+    val window = parseAdminStatsRange(range, clock) ?: return null
+    return get(window.first, window.second)
+}
+
+private fun parseAdminStatsRange(range: String?, clock: Clock): Pair<java.time.Instant, java.time.Instant>? {
     val days = when (range) {
         null, "30d" -> 30L
         "7d" -> 7L
@@ -507,7 +524,7 @@ private suspend fun AdminStatsService.getRange(
         else -> return null
     }
     val until = clock.instant()
-    return get(until.minus(Duration.ofDays(days)), until)
+    return until.minus(Duration.ofDays(days)) to until
 }
 
 private suspend fun ApplicationCall.requirePrincipal(
@@ -657,8 +674,6 @@ private fun AdminStatsDto.toReferralResponse(): AdminReferralResponse =
             AdminFunnelResponse("邀请码创建", referralFunnel.codesCreated),
             AdminFunnelResponse("成功绑定", referralFunnel.bindings),
             AdminFunnelResponse("有效使用并奖励", referralFunnel.rewardedBindings),
-            AdminFunnelResponse("待资格确认", referralFunnel.pendingBindings),
-            AdminFunnelResponse("不符合奖励条件", referralFunnel.ineligibleBindings),
         ),
         ranking = referralRanking.map {
             AdminReferralRankResponse(

src/main/kotlin/com/osglab/account/features/admin/stats/models/AdminProductAnalyticsDtos.kt

@@ -0,0 +1,113 @@
+package com.osglab.account.features.admin.stats.models
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class AdminAnalyticsRateDto(
+    val numerator: Long,
+    val denominator: Long,
+    val percent: Double?,
+)
+
+@Serializable
+data class AdminAnalyticsChannelDto(
+    val channel: String,
+    val installations: Long,
+    val activated: Long,
+    val activationRate: AdminAnalyticsRateDto,
+)
+
+@Serializable
+data class AdminAnalyticsCohortDto(
+    val cohortDate: String,
+    val size: Long,
+    val d1: AdminAnalyticsRateDto?,
+    val d7: AdminAnalyticsRateDto?,
+    val d30: AdminAnalyticsRateDto?,
+)
+
+@Serializable
+data class AdminAnalyticsFeatureUsageDto(
+    val feature: String,
+    val executionMode: String,
+    val users: Long,
+    val successes: Long,
+)
+
+@Serializable
+data class AdminAnalyticsFunnelStepDto(
+    val label: String,
+    val count: Long,
+)
+
+@Serializable
+data class AdminAnalyticsPeriodDto(
+    val from: String,
+    val until: String,
+)
+
+@Serializable
+data class AdminAnalyticsNorthStarDto(
+    val weeklyAiActiveUsers: Long,
+    val previousWeeklyAiActiveUsers: Long,
+    val weekOverWeekPercent: Double?,
+)
+
+@Serializable
+data class AdminAnalyticsGrowthDto(
+    val newInstallations: Long,
+    val newAccounts: Long,
+    val activation24h: AdminAnalyticsRateDto,
+    val medianTimeToValueMinutes: Double?,
+    val channels: List<AdminAnalyticsChannelDto>,
+)
+
+@Serializable
+data class AdminAnalyticsActivityDto(
+    val dau: Long,
+    val wau: Long,
+    val mau: Long,
+    val stickinessPercent: Double?,
+    val successfulAiRequests: Long,
+    val successfulRequestsPerActiveUser: Double?,
+)
+
+@Serializable
+data class AdminAnalyticsConsumptionDto(
+    val totalCredits: Long,
+    val averageDailyCreditsPerActiveUser: Double?,
+    val medianUserDailyCredits: Double?,
+    val averageCreditsPerManagedRequest: Double?,
+)
+
+@Serializable
+data class AdminAnalyticsMonetizationDto(
+    val payingUsers: Long,
+    val purchases: Long,
+    val creditsPurchased: Long,
+    val conversion7d: AdminAnalyticsRateDto,
+    val conversion30d: AdminAnalyticsRateDto,
+    val repeatPurchaseRate: AdminAnalyticsRateDto,
+)
+
+@Serializable
+data class AdminAnalyticsGuardrailsDto(
+    val clientAiSuccessRate: AdminAnalyticsRateDto,
+    val managedSuccessRate: AdminAnalyticsRateDto,
+    val creditBlockedUsers: Long,
+)
+
+@Serializable
+data class AdminProductAnalyticsDto(
+    val period: AdminAnalyticsPeriodDto,
+    val northStar: AdminAnalyticsNorthStarDto,
+    val growth: AdminAnalyticsGrowthDto,
+    val activity: AdminAnalyticsActivityDto,
+    val consumption: AdminAnalyticsConsumptionDto,
+    val monetization: AdminAnalyticsMonetizationDto,
+    val growthFunnel: List<AdminAnalyticsFunnelStepDto>,
+    val retention: List<AdminAnalyticsCohortDto>,
+    val aiFeatures: List<AdminAnalyticsFeatureUsageDto>,
+    val referralFunnel: List<AdminAnalyticsFunnelStepDto>,
+    val guardrails: AdminAnalyticsGuardrailsDto,
+)

src/main/kotlin/com/osglab/account/features/admin/stats/repositories/AdminProductAnalyticsRepository.kt

@@ -0,0 +1,829 @@
+package com.osglab.account.features.admin.stats.repositories
+
+import com.osglab.account.config.DatabaseFactory
+import org.jetbrains.exposed.v1.core.IColumnType
+import org.jetbrains.exposed.v1.javatime.JavaInstantColumnType
+import org.jetbrains.exposed.v1.jdbc.transactions.TransactionManager
+import java.math.BigDecimal
+import java.sql.ResultSet
+import java.time.Instant
+import java.time.LocalDate
+
+data class AdminAnalyticsWindow(
+    val from: Instant,
+    val until: Instant,
+) {
+    init {
+        require(from < until)
+    }
+}
+
+data class AdminAnalyticsCountRow(
+    val numerator: Long,
+    val denominator: Long,
+)
+
+data class AdminAnalyticsChannelRow(
+    val channel: String,
+    val installations: Long,
+    val activated: Long,
+)
+
+data class AdminAnalyticsCohortRow(
+    val cohortDate: LocalDate,
+    val size: Long,
+    val d1: Long,
+    val d7: Long,
+    val d30: Long,
+)
+
+data class AdminAnalyticsFeatureRow(
+    val feature: String,
+    val executionMode: String,
+    val users: Long,
+    val successes: Long,
+)
+
+data class AdminAnalyticsConsumptionRow(
+    val totalCredits: Long,
+    val managedRequests: Long,
+    val averageDailyCreditsPerActiveUser: Double?,
+    val medianUserDailyCredits: Double?,
+)
+
+data class AdminAnalyticsMonetizationRow(
+    val payingUsers: Long,
+    val purchases: Long,
+    val creditsPurchased: Long,
+    val conversion7d: AdminAnalyticsCountRow,
+    val conversion30d: AdminAnalyticsCountRow,
+    val repeatPurchase: AdminAnalyticsCountRow,
+)
+
+data class AdminAnalyticsReferralRow(
+    val shared: Long,
+    val opened: Long,
+    val bound: Long,
+    val activated: Long,
+    val rewarded: Long,
+)
+
+data class AdminAnalyticsGuardrailRow(
+    val clientSuccess: AdminAnalyticsCountRow,
+    val managedSuccess: AdminAnalyticsCountRow,
+    val creditBlockedUsers: Long,
+)
+
+data class AdminAnalyticsGrowthFunnelRow(
+    val opened: Long,
+    val registered: Long,
+    val activated: Long,
+    val retainedD7: Long,
+    val purchased: Long,
+)
+
+data class AdminProductAnalyticsSnapshot(
+    val currentWeeklyUsers: Long,
+    val previousWeeklyUsers: Long,
+    val newInstallations: Long,
+    val newAccounts: Long,
+    val activation24h: AdminAnalyticsCountRow,
+    val medianTimeToValueMinutes: Double?,
+    val channels: List<AdminAnalyticsChannelRow>,
+    val dau: Long,
+    val wau: Long,
+    val mau: Long,
+    val periodActiveUsers: Long,
+    val successfulAiRequests: Long,
+    val consumption: AdminAnalyticsConsumptionRow,
+    val monetization: AdminAnalyticsMonetizationRow,
+    val growthFunnel: AdminAnalyticsGrowthFunnelRow,
+    val retention: List<AdminAnalyticsCohortRow>,
+    val features: List<AdminAnalyticsFeatureRow>,
+    val referrals: AdminAnalyticsReferralRow,
+    val guardrails: AdminAnalyticsGuardrailRow,
+)
+
+interface AdminProductAnalyticsRepository {
+    suspend fun load(
+        range: AdminAnalyticsWindow,
+        currentWeek: AdminAnalyticsWindow,
+        previousWeek: AdminAnalyticsWindow,
+    ): AdminProductAnalyticsSnapshot
+}
+
+class ExposedAdminProductAnalyticsRepository(
+    private val databaseFactory: DatabaseFactory,
+) : AdminProductAnalyticsRepository {
+    override suspend fun load(
+        range: AdminAnalyticsWindow,
+        currentWeek: AdminAnalyticsWindow,
+        previousWeek: AdminAnalyticsWindow,
+    ): AdminProductAnalyticsSnapshot = databaseFactory.query {
+        val activation = loadActivation(range)
+        AdminProductAnalyticsSnapshot(
+            currentWeeklyUsers = loadValueActiveUsers(currentWeek),
+            previousWeeklyUsers = loadValueActiveUsers(previousWeek),
+            newInstallations = activation.denominator,
+            newAccounts = loadNewAccounts(range),
+            activation24h = AdminAnalyticsCountRow(activation.activated, activation.denominator),
+            medianTimeToValueMinutes = activation.medianMinutes,
+            channels = loadChannels(range),
+            dau = loadValueActiveUsers(
+                AdminAnalyticsWindow(range.until.minusSeconds(DAY_SECONDS), range.until),
+            ),
+            wau = loadValueActiveUsers(
+                AdminAnalyticsWindow(range.until.minusSeconds(7 * DAY_SECONDS), range.until),
+            ),
+            mau = loadValueActiveUsers(
+                AdminAnalyticsWindow(range.until.minusSeconds(30 * DAY_SECONDS), range.until),
+            ),
+            periodActiveUsers = loadValueActiveUsers(range),
+            successfulAiRequests = loadSuccessfulAiRequests(range),
+            consumption = loadConsumption(range),
+            monetization = loadMonetization(range),
+            growthFunnel = loadGrowthFunnel(range),
+            retention = loadRetention(range),
+            features = loadFeatures(range),
+            referrals = loadReferrals(range),
+            guardrails = loadGuardrails(range),
+        )
+    }
+
+    private fun loadValueActiveUsers(window: AdminAnalyticsWindow): Long =
+        querySingle(
+            valueEventsCte() +
+                """
+                SELECT COUNT(DISTINCT identity_key) AS aggregate_value
+                FROM value_events
+                WHERE occurred_at >= ? AND occurred_at < ?
+                """,
+            window.arguments(),
+        ) { it.exactLong("aggregate_value") }
+
+    private fun loadNewAccounts(range: AdminAnalyticsWindow): Long =
+        querySingle(
+            """
+            SELECT COUNT(*) AS aggregate_value
+            FROM accounts
+            WHERE created_at >= ? AND created_at < ?
+            """,
+            range.arguments(),
+        ) { it.exactLong("aggregate_value") }
+
+    private fun loadActivation(range: AdminAnalyticsWindow): ActivationRow =
+        querySingle(
+            """
+            WITH first_open AS (
+                SELECT installation_hash, MIN(occurred_at) AS opened_at
+                FROM product_analytics_events
+                WHERE event_name = 'FIRST_OPEN'
+                  AND occurred_at >= ? AND occurred_at < ?
+                GROUP BY installation_hash
+            ),
+            client_value AS (
+                SELECT installation_hash, MIN(occurred_at) AS value_at
+                FROM product_analytics_events
+                WHERE event_name = 'AI_FEATURE_SUCCEEDED'
+                  AND execution_mode IN ('LOCAL', 'BYOK')
+                GROUP BY installation_hash
+            ),
+            managed_value AS (
+                SELECT i.installation_hash, MIN(u.created_at) AS value_at
+                FROM product_analytics_installations i
+                JOIN credit_usage_records u ON u.user_id = i.account_id
+                GROUP BY i.installation_hash
+            ),
+            first_value_by_install AS (
+                SELECT installation_hash, MIN(value_at) AS value_at
+                FROM (
+                    SELECT * FROM client_value
+                    UNION ALL
+                    SELECT * FROM managed_value
+                ) values_by_source
+                GROUP BY installation_hash
+            ),
+            activated AS (
+                SELECT
+                    o.installation_hash,
+                    TIMESTAMPDIFF(SECOND, o.opened_at, v.value_at) AS seconds_to_value
+                FROM first_open o
+                JOIN first_value_by_install v ON v.installation_hash = o.installation_hash
+                WHERE v.value_at >= o.opened_at
+                  AND v.value_at <= DATE_ADD(o.opened_at, INTERVAL 24 HOUR)
+            ),
+            ranked AS (
+                SELECT
+                    seconds_to_value,
+                    ROW_NUMBER() OVER (ORDER BY seconds_to_value) AS row_number_value,
+                    COUNT(*) OVER () AS total_rows
+                FROM activated
+            )
+            SELECT
+                (SELECT COUNT(*) FROM first_open) AS denominator_value,
+                (SELECT COUNT(*) FROM activated) AS activated_value,
+                (
+                    SELECT AVG(seconds_to_value) / 60.0
+                    FROM ranked
+                    WHERE row_number_value IN (
+                        FLOOR((total_rows + 1) / 2),
+                        FLOOR((total_rows + 2) / 2)
+                    )
+                ) AS median_minutes
+            """,
+            range.arguments(),
+        ) {
+            ActivationRow(
+                denominator = it.exactLong("denominator_value"),
+                activated = it.exactLong("activated_value"),
+                medianMinutes = it.nullableDouble("median_minutes"),
+            )
+        }
+
+    private fun loadChannels(range: AdminAnalyticsWindow): List<AdminAnalyticsChannelRow> =
+        queryRows(
+            """
+            WITH first_open AS (
+                SELECT
+                    e.installation_hash,
+                    MIN(e.occurred_at) AS opened_at,
+                    COALESCE(
+                        MAX(
+                            CASE
+                                WHEN e.acquisition_channel <> 'UNKNOWN'
+                                THEN e.acquisition_channel
+                            END
+                        ),
+                        'UNKNOWN'
+                    ) AS channel
+                FROM product_analytics_events e
+                WHERE e.event_name = 'FIRST_OPEN'
+                  AND e.occurred_at >= ? AND e.occurred_at < ?
+                GROUP BY e.installation_hash
+            ),
+            client_value AS (
+                SELECT installation_hash, MIN(occurred_at) AS value_at
+                FROM product_analytics_events
+                WHERE event_name = 'AI_FEATURE_SUCCEEDED'
+                  AND execution_mode IN ('LOCAL', 'BYOK')
+                GROUP BY installation_hash
+            ),
+            managed_value AS (
+                SELECT i.installation_hash, MIN(u.created_at) AS value_at
+                FROM product_analytics_installations i
+                JOIN credit_usage_records u ON u.user_id = i.account_id
+                GROUP BY i.installation_hash
+            ),
+            first_value_by_install AS (
+                SELECT installation_hash, MIN(value_at) AS value_at
+                FROM (
+                    SELECT * FROM client_value
+                    UNION ALL
+                    SELECT * FROM managed_value
+                ) values_by_source
+                GROUP BY installation_hash
+            )
+            SELECT
+                o.channel,
+                COUNT(*) AS installations,
+                SUM(
+                    CASE
+                        WHEN v.value_at >= o.opened_at
+                         AND v.value_at <= DATE_ADD(o.opened_at, INTERVAL 24 HOUR)
+                        THEN 1 ELSE 0
+                    END
+                ) AS activated
+            FROM first_open o
+            LEFT JOIN first_value_by_install v ON v.installation_hash = o.installation_hash
+            GROUP BY o.channel
+            ORDER BY o.channel
+            """,
+            range.arguments(),
+        ) {
+            AdminAnalyticsChannelRow(
+                channel = it.getString("channel"),
+                installations = it.exactLong("installations"),
+                activated = it.exactLong("activated"),
+            )
+        }
+
+    private fun loadSuccessfulAiRequests(range: AdminAnalyticsWindow): Long =
+        querySingle(
+            """
+            SELECT
+                (
+                    SELECT COUNT(*)
+                    FROM credit_usage_records
+                    WHERE created_at >= ? AND created_at < ?
+                ) + (
+                    SELECT COUNT(*)
+                    FROM product_analytics_events
+                    WHERE occurred_at >= ? AND occurred_at < ?
+                      AND event_name = 'AI_FEATURE_SUCCEEDED'
+                      AND execution_mode IN ('LOCAL', 'BYOK')
+                ) AS aggregate_value
+            """,
+            range.arguments(repetitions = 2),
+        ) { it.exactLong("aggregate_value") }
+
+    private fun loadConsumption(range: AdminAnalyticsWindow): AdminAnalyticsConsumptionRow =
+        querySingle(
+            """
+            WITH user_days AS (
+                SELECT
+                    user_id,
+                    DATE(created_at) AS usage_date,
+                    SUM(charged_credits) AS daily_credits
+                FROM credit_usage_records
+                WHERE created_at >= ? AND created_at < ?
+                GROUP BY user_id, DATE(created_at)
+            ),
+            day_totals AS (
+                SELECT
+                    usage_date,
+                    SUM(daily_credits) AS credits,
+                    COUNT(*) AS users
+                FROM user_days
+                GROUP BY usage_date
+            ),
+            ranked_user_days AS (
+                SELECT
+                    daily_credits,
+                    ROW_NUMBER() OVER (ORDER BY daily_credits) AS row_number_value,
+                    COUNT(*) OVER () AS total_rows
+                FROM user_days
+            )
+            SELECT
+                COALESCE((SELECT SUM(daily_credits) FROM user_days), 0) AS total_credits,
+                COALESCE((SELECT COUNT(*) FROM credit_usage_records
+                          WHERE created_at >= ? AND created_at < ?), 0) AS managed_requests,
+                (SELECT AVG(credits / NULLIF(users, 0)) FROM day_totals)
+                    AS average_daily_per_user,
+                (
+                    SELECT AVG(daily_credits)
+                    FROM ranked_user_days
+                    WHERE row_number_value IN (
+                        FLOOR((total_rows + 1) / 2),
+                        FLOOR((total_rows + 2) / 2)
+                    )
+                ) AS median_user_day
+            """,
+            range.arguments(repetitions = 2),
+        ) {
+            AdminAnalyticsConsumptionRow(
+                totalCredits = it.exactLong("total_credits"),
+                managedRequests = it.exactLong("managed_requests"),
+                averageDailyCreditsPerActiveUser = it.nullableDouble("average_daily_per_user"),
+                medianUserDailyCredits = it.nullableDouble("median_user_day"),
+            )
+        }
+
+    private fun loadMonetization(range: AdminAnalyticsWindow): AdminAnalyticsMonetizationRow {
+        val sevenDayMaturity = range.until.minusSeconds(7 * DAY_SECONDS)
+        val thirtyDayMaturity = range.until.minusSeconds(30 * DAY_SECONDS)
+        return querySingle(
+            """
+            WITH first_purchase AS (
+                SELECT user_id, MIN(purchased_at) AS first_purchased_at, COUNT(*) AS lifetime_purchases
+                FROM storekit_credit_purchases
+                GROUP BY user_id
+            ),
+            period_payers AS (
+                SELECT user_id, COUNT(*) AS period_purchases
+                FROM storekit_credit_purchases
+                WHERE purchased_at >= ? AND purchased_at < ?
+                GROUP BY user_id
+            )
+            SELECT
+                (SELECT COUNT(*) FROM period_payers) AS paying_users,
+                (
+                    SELECT COUNT(*)
+                    FROM storekit_credit_purchases
+                    WHERE purchased_at >= ? AND purchased_at < ?
+                ) AS purchases,
+                (
+                    SELECT COALESCE(SUM(credits_granted), 0)
+                    FROM storekit_credit_purchases
+                    WHERE purchased_at >= ? AND purchased_at < ?
+                ) AS credits_purchased,
+                (
+                    SELECT COUNT(*)
+                    FROM accounts
+                    WHERE created_at >= ? AND created_at < ?
+                ) AS conversion_7_denominator,
+                (
+                    SELECT COUNT(*)
+                    FROM accounts a
+                    JOIN first_purchase p ON p.user_id = a.id
+                    WHERE a.created_at >= ? AND a.created_at < ?
+                      AND p.first_purchased_at <= DATE_ADD(a.created_at, INTERVAL 7 DAY)
+                ) AS conversion_7_numerator,
+                (
+                    SELECT COUNT(*)
+                    FROM accounts
+                    WHERE created_at >= ? AND created_at < ?
+                ) AS conversion_30_denominator,
+                (
+                    SELECT COUNT(*)
+                    FROM accounts a
+                    JOIN first_purchase p ON p.user_id = a.id
+                    WHERE a.created_at >= ? AND a.created_at < ?
+                      AND p.first_purchased_at <= DATE_ADD(a.created_at, INTERVAL 30 DAY)
+                ) AS conversion_30_numerator,
+                (
+                    SELECT COUNT(*)
+                    FROM period_payers pp
+                    JOIN first_purchase fp ON fp.user_id = pp.user_id
+                    WHERE fp.lifetime_purchases >= 2
+                ) AS repeat_numerator,
+                (SELECT COUNT(*) FROM period_payers) AS repeat_denominator
+            """,
+            buildList {
+                addAll(range.arguments(repetitions = 3))
+                addAll(maturedWindowArguments(range.from, sevenDayMaturity))
+                addAll(maturedWindowArguments(range.from, sevenDayMaturity))
+                addAll(maturedWindowArguments(range.from, thirtyDayMaturity))
+                addAll(maturedWindowArguments(range.from, thirtyDayMaturity))
+            },
+        ) {
+            AdminAnalyticsMonetizationRow(
+                payingUsers = it.exactLong("paying_users"),
+                purchases = it.exactLong("purchases"),
+                creditsPurchased = it.exactLong("credits_purchased"),
+                conversion7d = AdminAnalyticsCountRow(
+                    it.exactLong("conversion_7_numerator"),
+                    it.exactLong("conversion_7_denominator"),
+                ),
+                conversion30d = AdminAnalyticsCountRow(
+                    it.exactLong("conversion_30_numerator"),
+                    it.exactLong("conversion_30_denominator"),
+                ),
+                repeatPurchase = AdminAnalyticsCountRow(
+                    it.exactLong("repeat_numerator"),
+                    it.exactLong("repeat_denominator"),
+                ),
+            )
+        }
+    }
+
+    private fun loadGrowthFunnel(range: AdminAnalyticsWindow): AdminAnalyticsGrowthFunnelRow =
+        querySingle(
+            """
+            WITH first_open AS (
+                SELECT installation_hash, MIN(occurred_at) AS opened_at
+                FROM product_analytics_events
+                WHERE event_name = 'FIRST_OPEN'
+                  AND occurred_at >= ? AND occurred_at < ?
+                GROUP BY installation_hash
+            ),
+            client_values AS (
+                SELECT installation_hash, occurred_at
+                FROM product_analytics_events
+                WHERE event_name = 'AI_FEATURE_SUCCEEDED'
+                  AND execution_mode IN ('LOCAL', 'BYOK')
+            ),
+            managed_values AS (
+                SELECT i.installation_hash, u.created_at AS occurred_at
+                FROM product_analytics_installations i
+                JOIN credit_usage_records u ON u.user_id = i.account_id
+            ),
+            values_by_install AS (
+                SELECT * FROM client_values
+                UNION ALL
+                SELECT * FROM managed_values
+            ),
+            activated AS (
+                SELECT
+                    o.installation_hash,
+                    o.opened_at,
+                    MIN(v.occurred_at) AS first_value_at
+                FROM first_open o
+                JOIN values_by_install v ON v.installation_hash = o.installation_hash
+                WHERE v.occurred_at >= o.opened_at
+                  AND v.occurred_at <= DATE_ADD(o.opened_at, INTERVAL 24 HOUR)
+                GROUP BY o.installation_hash, o.opened_at
+            )
+            SELECT
+                (SELECT COUNT(*) FROM first_open) AS opened,
+                (
+                    SELECT COUNT(*)
+                    FROM first_open o
+                    JOIN product_analytics_installations i
+                      ON i.installation_hash = o.installation_hash
+                    WHERE i.account_id IS NOT NULL
+                ) AS registered,
+                (SELECT COUNT(*) FROM activated) AS activated,
+                (
+                    SELECT COUNT(*)
+                    FROM activated a
+                    WHERE EXISTS (
+                        SELECT 1
+                        FROM values_by_install v
+                        WHERE v.installation_hash = a.installation_hash
+                          AND DATE(v.occurred_at) = DATE_ADD(DATE(a.first_value_at), INTERVAL 7 DAY)
+                    )
+                ) AS retained_d7,
+                (
+                    SELECT COUNT(*)
+                    FROM first_open o
+                    JOIN product_analytics_installations i
+                      ON i.installation_hash = o.installation_hash
+                    WHERE EXISTS (
+                        SELECT 1
+                        FROM storekit_credit_purchases p
+                        WHERE p.user_id = i.account_id
+                          AND p.purchased_at >= o.opened_at
+                          AND p.purchased_at < ?
+                    )
+                ) AS purchased
+            """,
+            range.arguments() + listOf(INSTANT_COLUMN_TYPE to range.until),
+        ) {
+            AdminAnalyticsGrowthFunnelRow(
+                opened = it.exactLong("opened"),
+                registered = it.exactLong("registered"),
+                activated = it.exactLong("activated"),
+                retainedD7 = it.exactLong("retained_d7"),
+                purchased = it.exactLong("purchased"),
+            )
+        }
+
+    private fun loadRetention(range: AdminAnalyticsWindow): List<AdminAnalyticsCohortRow> =
+        queryRows(
+            valueEventsCte() +
+                """
+                , first_value_by_identity AS (
+                    SELECT identity_key, MIN(occurred_at) AS first_value_at
+                    FROM value_events
+                    GROUP BY identity_key
+                )
+                SELECT
+                    DATE(f.first_value_at) AS cohort_date,
+                    COUNT(DISTINCT f.identity_key) AS cohort_size,
+                    COUNT(
+                        DISTINCT CASE
+                            WHEN DATEDIFF(DATE(v.occurred_at), DATE(f.first_value_at)) = 1
+                            THEN f.identity_key
+                        END
+                    ) AS retained_d1,
+                    COUNT(
+                        DISTINCT CASE
+                            WHEN DATEDIFF(DATE(v.occurred_at), DATE(f.first_value_at)) = 7
+                            THEN f.identity_key
+                        END
+                    ) AS retained_d7,
+                    COUNT(
+                        DISTINCT CASE
+                            WHEN DATEDIFF(DATE(v.occurred_at), DATE(f.first_value_at)) = 30
+                            THEN f.identity_key
+                        END
+                    ) AS retained_d30
+                FROM first_value_by_identity f
+                LEFT JOIN value_events v ON v.identity_key = f.identity_key
+                WHERE f.first_value_at >= ? AND f.first_value_at < ?
+                GROUP BY DATE(f.first_value_at)
+                ORDER BY cohort_date DESC
+                """,
+            range.arguments(),
+        ) {
+            AdminAnalyticsCohortRow(
+                cohortDate = it.getObject("cohort_date", LocalDate::class.java),
+                size = it.exactLong("cohort_size"),
+                d1 = it.exactLong("retained_d1"),
+                d7 = it.exactLong("retained_d7"),
+                d30 = it.exactLong("retained_d30"),
+            )
+        }
+
+    private fun loadFeatures(range: AdminAnalyticsWindow): List<AdminAnalyticsFeatureRow> =
+        queryRows(
+            """
+            SELECT
+                e.feature,
+                e.execution_mode,
+                COUNT(
+                    DISTINCT COALESCE(
+                        CONCAT('a:', i.account_id),
+                        CONCAT('i:', e.installation_hash)
+                    )
+                ) AS users,
+                COUNT(*) AS successes
+            FROM product_analytics_events e
+            JOIN product_analytics_installations i
+              ON i.installation_hash = e.installation_hash
+            WHERE e.occurred_at >= ? AND e.occurred_at < ?
+              AND e.event_name = 'AI_FEATURE_SUCCEEDED'
+            GROUP BY e.feature, e.execution_mode
+            ORDER BY successes DESC, e.feature, e.execution_mode
+            """,
+            range.arguments(),
+        ) {
+            AdminAnalyticsFeatureRow(
+                feature = it.getString("feature"),
+                executionMode = it.getString("execution_mode"),
+                users = it.exactLong("users"),
+                successes = it.exactLong("successes"),
+            )
+        }
+
+    private fun loadReferrals(range: AdminAnalyticsWindow): AdminAnalyticsReferralRow =
+        querySingle(
+            valueEventsCte() +
+                """
+                SELECT
+                    (
+                        SELECT COUNT(DISTINCT installation_hash)
+                        FROM product_analytics_events
+                        WHERE event_name = 'REFERRAL_SHARED'
+                          AND occurred_at >= ? AND occurred_at < ?
+                    ) AS shared,
+                    (
+                        SELECT COUNT(DISTINCT installation_hash)
+                        FROM product_analytics_events
+                        WHERE event_name = 'INVITE_OPENED'
+                          AND occurred_at >= ? AND occurred_at < ?
+                    ) + (
+                        SELECT COALESCE(SUM(counter_value), 0)
+                        FROM product_analytics_daily_counters
+                        WHERE counter_name = 'INVITE_PAGE_OPENED'
+                          AND counter_date >= DATE(?) AND counter_date < DATE(?)
+                    ) AS opened,
+                    (
+                        SELECT COUNT(*)
+                        FROM referral_bindings
+                        WHERE bound_at >= ? AND bound_at < ?
+                    ) AS bound,
+                    (
+                        SELECT COUNT(*)
+                        FROM referral_bindings r
+                        WHERE r.bound_at >= ? AND r.bound_at < ?
+                          AND EXISTS (
+                              SELECT 1
+                              FROM value_events v
+                              WHERE v.identity_key = CONCAT('a:', r.invitee_user_id)
+                                AND v.occurred_at >= r.bound_at
+                                AND v.occurred_at < ?
+                          )
+                    ) AS activated,
+                    (
+                        SELECT COUNT(*)
+                        FROM referral_bindings
+                        WHERE bound_at >= ? AND bound_at < ?
+                          AND reward_status = 'REWARDED'
+                    ) AS rewarded
+                """,
+            buildList {
+                addAll(range.arguments(repetitions = 5))
+                add(INSTANT_COLUMN_TYPE to range.until)
+                addAll(range.arguments())
+            },
+        ) {
+            AdminAnalyticsReferralRow(
+                shared = it.exactLong("shared"),
+                opened = it.exactLong("opened"),
+                bound = it.exactLong("bound"),
+                activated = it.exactLong("activated"),
+                rewarded = it.exactLong("rewarded"),
+            )
+        }
+
+    private fun loadGuardrails(range: AdminAnalyticsWindow): AdminAnalyticsGuardrailRow =
+        querySingle(
+            """
+            SELECT
+                (
+                    SELECT COUNT(*)
+                    FROM product_analytics_events
+                    WHERE occurred_at >= ? AND occurred_at < ?
+                      AND event_name = 'AI_FEATURE_SUCCEEDED'
+                ) AS client_success,
+                (
+                    SELECT COUNT(*)
+                    FROM product_analytics_events
+                    WHERE occurred_at >= ? AND occurred_at < ?
+                      AND event_name IN ('AI_FEATURE_SUCCEEDED', 'AI_FEATURE_FAILED')
+                ) AS client_terminal,
+                (
+                    SELECT COUNT(*)
+                    FROM provider_requests
+                    WHERE completed_at >= ? AND completed_at < ?
+                      AND status = 'SETTLED'
+                ) AS managed_success,
+                (
+                    SELECT COUNT(*)
+                    FROM provider_requests
+                    WHERE completed_at >= ? AND completed_at < ?
+                      AND status IN ('SETTLED', 'RELEASED', 'MANUAL_REVIEW')
+                ) AS managed_terminal,
+                (
+                    SELECT COUNT(
+                        DISTINCT COALESCE(
+                            CONCAT('a:', i.account_id),
+                            CONCAT('i:', e.installation_hash)
+                        )
+                    )
+                    FROM product_analytics_events e
+                    JOIN product_analytics_installations i
+                      ON i.installation_hash = e.installation_hash
+                    WHERE e.occurred_at >= ? AND e.occurred_at < ?
+                      AND e.event_name = 'AI_FEATURE_FAILED'
+                      AND e.failure_category = 'INSUFFICIENT_CREDITS'
+                ) AS credit_blocked_users
+            """,
+            range.arguments(repetitions = 5),
+        ) {
+            AdminAnalyticsGuardrailRow(
+                clientSuccess = AdminAnalyticsCountRow(
+                    it.exactLong("client_success"),
+                    it.exactLong("client_terminal"),
+                ),
+                managedSuccess = AdminAnalyticsCountRow(
+                    it.exactLong("managed_success"),
+                    it.exactLong("managed_terminal"),
+                ),
+                creditBlockedUsers = it.exactLong("credit_blocked_users"),
+            )
+        }
+}
+
+private data class ActivationRow(
+    val denominator: Long,
+    val activated: Long,
+    val medianMinutes: Double?,
+)
+
+private fun valueEventsCte(): String =
+    """
+    WITH value_events AS (
+        SELECT
+            CONCAT('a:', user_id) AS identity_key,
+            created_at AS occurred_at
+        FROM credit_usage_records
+        UNION ALL
+        SELECT
+            COALESCE(
+                CONCAT('a:', i.account_id),
+                CONCAT('i:', e.installation_hash)
+            ) AS identity_key,
+            e.occurred_at
+        FROM product_analytics_events e
+        JOIN product_analytics_installations i
+          ON i.installation_hash = e.installation_hash
+        WHERE e.event_name = 'AI_FEATURE_SUCCEEDED'
+          AND e.execution_mode IN ('LOCAL', 'BYOK')
+    )
+    """.trimIndent()
+
+private fun AdminAnalyticsWindow.arguments(
+    repetitions: Int = 1,
+): List<Pair<IColumnType<*>, Any?>> = buildList {
+    repeat(repetitions) {
+        add(INSTANT_COLUMN_TYPE to from)
+        add(INSTANT_COLUMN_TYPE to until)
+    }
+}
+
+private fun maturedWindowArguments(
+    from: Instant,
+    maturityEnd: Instant,
+): List<Pair<IColumnType<*>, Any?>> =
+    listOf(
+        INSTANT_COLUMN_TYPE to from,
+        INSTANT_COLUMN_TYPE to maxOf(from, maturityEnd),
+    )
+
+private fun <T> querySingle(
+    sql: String,
+    arguments: List<Pair<IColumnType<*>, Any?>>,
+    transform: (ResultSet) -> T,
+): T = queryRows(sql, arguments, transform).single()
+
+private fun <T> queryRows(
+    sql: String,
+    arguments: List<Pair<IColumnType<*>, Any?>>,
+    transform: (ResultSet) -> T,
+): List<T> {
+    val normalized = sql.trimIndent()
+    // Exposed classifies statements beginning with WITH as updates. Wrapping the
+    // CTE keeps prepared arguments and makes the statement unambiguously a query.
+    val executable = if (normalized.startsWith("WITH ", ignoreCase = true)) {
+        "SELECT * FROM (\n$normalized\n) AS analytics_result"
+    } else {
+        normalized
+    }
+    return TransactionManager.current().exec(executable, arguments) { result ->
+    buildList {
+        while (result.next()) add(transform(result))
+    }
+} ?: emptyList()
+}
+
+private fun ResultSet.exactLong(column: String): Long =
+    requireNotNull(getBigDecimal(column)) { "Aggregate column $column must not be null" }
+        .longValueExact()
+
+private fun ResultSet.nullableDouble(column: String): Double? =
+    getBigDecimal(column)?.toDouble()
+
+private val INSTANT_COLUMN_TYPE = JavaInstantColumnType()
+private const val DAY_SECONDS = 86_400L

src/main/kotlin/com/osglab/account/features/admin/stats/services/AdminProductAnalyticsService.kt

@@ -0,0 +1,187 @@
+package com.osglab.account.features.admin.stats.services
+
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsActivityDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsChannelDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsCohortDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsConsumptionDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsFeatureUsageDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsFunnelStepDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsGrowthDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsGuardrailsDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsMonetizationDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsNorthStarDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsPeriodDto
+import com.osglab.account.features.admin.stats.models.AdminAnalyticsRateDto
+import com.osglab.account.features.admin.stats.models.AdminProductAnalyticsDto
+import com.osglab.account.features.admin.stats.repositories.AdminAnalyticsCountRow
+import com.osglab.account.features.admin.stats.repositories.AdminAnalyticsWindow
+import com.osglab.account.features.admin.stats.repositories.AdminProductAnalyticsRepository
+import java.math.BigDecimal
+import java.math.RoundingMode
+import java.time.DayOfWeek
+import java.time.Duration
+import java.time.Instant
+import java.time.ZoneOffset
+import java.time.temporal.TemporalAdjusters
+
+class AdminProductAnalyticsService(
+    private val repository: AdminProductAnalyticsRepository,
+) {
+    suspend fun get(from: Instant, until: Instant): AdminProductAnalyticsDto {
+        require(from < until)
+        val currentWeek = currentWeekWindow(until)
+        val elapsed = Duration.between(currentWeek.from, currentWeek.until)
+        val previousWeek = AdminAnalyticsWindow(
+            from = currentWeek.from.minus(Duration.ofDays(7)),
+            until = currentWeek.from.minus(Duration.ofDays(7)).plus(elapsed),
+        )
+        val snapshot = repository.load(
+            range = AdminAnalyticsWindow(from, until),
+            currentWeek = currentWeek,
+            previousWeek = previousWeek,
+        )
+        val referrals = snapshot.referrals
+        val growth = snapshot.growthFunnel
+        return AdminProductAnalyticsDto(
+            period = AdminAnalyticsPeriodDto(from.toString(), until.toString()),
+            northStar = AdminAnalyticsNorthStarDto(
+                weeklyAiActiveUsers = snapshot.currentWeeklyUsers,
+                previousWeeklyAiActiveUsers = snapshot.previousWeeklyUsers,
+                weekOverWeekPercent = growthPercent(
+                    snapshot.currentWeeklyUsers,
+                    snapshot.previousWeeklyUsers,
+                ),
+            ),
+            growth = AdminAnalyticsGrowthDto(
+                newInstallations = snapshot.newInstallations,
+                newAccounts = snapshot.newAccounts,
+                activation24h = snapshot.activation24h.toRate(),
+                medianTimeToValueMinutes = snapshot.medianTimeToValueMinutes?.rounded(),
+                channels = snapshot.channels.map {
+                    AdminAnalyticsChannelDto(
+                        channel = it.channel,
+                        installations = it.installations,
+                        activated = it.activated,
+                        activationRate = AdminAnalyticsCountRow(
+                            it.activated,
+                            it.installations,
+                        ).toRate(),
+                    )
+                },
+            ),
+            activity = AdminAnalyticsActivityDto(
+                dau = snapshot.dau,
+                wau = snapshot.wau,
+                mau = snapshot.mau,
+                stickinessPercent = percentage(snapshot.dau, snapshot.mau),
+                successfulAiRequests = snapshot.successfulAiRequests,
+                successfulRequestsPerActiveUser = ratio(
+                    snapshot.successfulAiRequests,
+                    snapshot.periodActiveUsers,
+                ),
+            ),
+            consumption = AdminAnalyticsConsumptionDto(
+                totalCredits = snapshot.consumption.totalCredits,
+                averageDailyCreditsPerActiveUser =
+                    snapshot.consumption.averageDailyCreditsPerActiveUser?.rounded(),
+                medianUserDailyCredits = snapshot.consumption.medianUserDailyCredits?.rounded(),
+                averageCreditsPerManagedRequest = ratio(
+                    snapshot.consumption.totalCredits,
+                    snapshot.consumption.managedRequests,
+                ),
+            ),
+            monetization = AdminAnalyticsMonetizationDto(
+                payingUsers = snapshot.monetization.payingUsers,
+                purchases = snapshot.monetization.purchases,
+                creditsPurchased = snapshot.monetization.creditsPurchased,
+                conversion7d = snapshot.monetization.conversion7d.toRate(),
+                conversion30d = snapshot.monetization.conversion30d.toRate(),
+                repeatPurchaseRate = snapshot.monetization.repeatPurchase.toRate(),
+            ),
+            growthFunnel = listOf(
+                AdminAnalyticsFunnelStepDto("首次启动", growth.opened),
+                AdminAnalyticsFunnelStepDto("完成注册", growth.registered),
+                AdminAnalyticsFunnelStepDto("24 小时内首次 AI 成功", growth.activated),
+                AdminAnalyticsFunnelStepDto("D7 再次使用 AI", growth.retainedD7),
+                AdminAnalyticsFunnelStepDto("首次购买", growth.purchased),
+            ),
+            retention = snapshot.retention.map { cohort ->
+                AdminAnalyticsCohortDto(
+                    cohortDate = cohort.cohortDate.toString(),
+                    size = cohort.size,
+                    d1 = cohort.takeIf { isMature(it.cohortDate.atStartOfDay().toInstant(ZoneOffset.UTC), 1, until) }
+                        ?.let { AdminAnalyticsCountRow(it.d1, it.size).toRate() },
+                    d7 = cohort.takeIf { isMature(it.cohortDate.atStartOfDay().toInstant(ZoneOffset.UTC), 7, until) }
+                        ?.let { AdminAnalyticsCountRow(it.d7, it.size).toRate() },
+                    d30 = cohort.takeIf {
+                        isMature(it.cohortDate.atStartOfDay().toInstant(ZoneOffset.UTC), 30, until)
+                    }?.let { AdminAnalyticsCountRow(it.d30, it.size).toRate() },
+                )
+            },
+            aiFeatures = snapshot.features.map {
+                AdminAnalyticsFeatureUsageDto(
+                    feature = it.feature,
+                    executionMode = it.executionMode,
+                    users = it.users,
+                    successes = it.successes,
+                )
+            },
+            referralFunnel = listOf(
+                AdminAnalyticsFunnelStepDto("发起分享", referrals.shared),
+                AdminAnalyticsFunnelStepDto("打开邀请", referrals.opened),
+                AdminAnalyticsFunnelStepDto("完成绑定", referrals.bound),
+                AdminAnalyticsFunnelStepDto("首次 AI 成功", referrals.activated),
+                AdminAnalyticsFunnelStepDto("完成奖励", referrals.rewarded),
+            ),
+            guardrails = AdminAnalyticsGuardrailsDto(
+                clientAiSuccessRate = snapshot.guardrails.clientSuccess.toRate(),
+                managedSuccessRate = snapshot.guardrails.managedSuccess.toRate(),
+                creditBlockedUsers = snapshot.guardrails.creditBlockedUsers,
+            ),
+        )
+    }
+}
+
+private fun currentWeekWindow(until: Instant): AdminAnalyticsWindow {
+    val referenceDate = until.minusNanos(1).atZone(ZoneOffset.UTC).toLocalDate()
+    val weekStart = referenceDate
+        .with(TemporalAdjusters.previousOrSame(DayOfWeek.MONDAY))
+        .atStartOfDay(ZoneOffset.UTC)
+        .toInstant()
+    return AdminAnalyticsWindow(weekStart, until)
+}
+
+private fun AdminAnalyticsCountRow.toRate(): AdminAnalyticsRateDto =
+    AdminAnalyticsRateDto(
+        numerator = numerator,
+        denominator = denominator,
+        percent = percentage(numerator, denominator),
+    )
+
+private fun percentage(numerator: Long, denominator: Long): Double? =
+    if (denominator == 0L) null else
+        numerator.toBigDecimal()
+            .multiply(HUNDRED)
+            .divide(denominator.toBigDecimal(), 1, RoundingMode.HALF_UP)
+            .toDouble()
+
+private fun ratio(numerator: Long, denominator: Long): Double? =
+    if (denominator == 0L) null else
+        numerator.toBigDecimal()
+            .divide(denominator.toBigDecimal(), 2, RoundingMode.HALF_UP)
+            .toDouble()
+
+private fun growthPercent(current: Long, previous: Long): Double? =
+    if (previous == 0L) null else
+        (current - previous).toBigDecimal()
+            .multiply(HUNDRED)
+            .divide(previous.toBigDecimal(), 1, RoundingMode.HALF_UP)
+            .toDouble()
+
+private fun Double.rounded(scale: Int = 1): Double =
+    toBigDecimal().setScale(scale, RoundingMode.HALF_UP).toDouble()
+
+private fun isMature(cohortStart: Instant, offsetDays: Long, until: Instant): Boolean =
+    !until.isBefore(cohortStart.plus(Duration.ofDays(offsetDays + 1)))
+
+private val HUNDRED = BigDecimal.valueOf(100)

src/main/kotlin/com/osglab/account/features/analytics/domain/AnalyticsDomain.kt

@@ -0,0 +1,113 @@
+package com.osglab.account.features.analytics.domain
+
+import com.osglab.account.common.errors.ApiException
+import io.ktor.http.HttpStatusCode
+import java.time.Instant
+import java.util.UUID
+import kotlinx.serialization.Serializable
+
+@Serializable
+enum class AnalyticsEventType {
+    FIRST_OPEN,
+    SESSION_STARTED,
+    KEYBOARD_ACTIVATED,
+    AI_FEATURE_STARTED,
+    AI_FEATURE_SUCCEEDED,
+    AI_FEATURE_FAILED,
+    PURCHASE_VIEWED,
+    PURCHASE_STARTED,
+    PURCHASE_CANCELLED,
+    REFERRAL_SHARED,
+    INVITE_OPENED,
+}
+
+@Serializable
+enum class AnalyticsSurface {
+    APP,
+    KEYBOARD,
+    INVITE_WEB,
+}
+
+@Serializable
+enum class AnalyticsAcquisitionChannel {
+    APP_STORE_ORGANIC,
+    REFERRAL,
+    SOCIAL_CONTENT,
+    UNKNOWN,
+}
+
+@Serializable
+enum class AnalyticsFeature {
+    TRANSCRIPTION,
+    POLISH,
+    AI_ASSISTANT,
+    AGENT,
+    HOTWORD,
+    OTHER,
+}
+
+@Serializable
+enum class AnalyticsExecutionMode {
+    MANAGED,
+    LOCAL,
+    BYOK,
+}
+
+@Serializable
+enum class AnalyticsFailureCategory {
+    NETWORK,
+    PROVIDER,
+    TIMEOUT,
+    CANCELLED,
+    INSUFFICIENT_CREDITS,
+    VALIDATION,
+    UNKNOWN,
+}
+
+@Serializable
+enum class AnalyticsDurationBucket {
+    LT_1S,
+    S1_TO_3,
+    S3_TO_10,
+    S10_TO_30,
+    GTE_30S,
+}
+
+data class AnalyticsEvent(
+    val clientEventId: UUID,
+    val eventType: AnalyticsEventType,
+    val occurredAt: Instant,
+    val surface: AnalyticsSurface,
+    val acquisitionChannel: AnalyticsAcquisitionChannel?,
+    val feature: AnalyticsFeature?,
+    val executionMode: AnalyticsExecutionMode?,
+    val failureCategory: AnalyticsFailureCategory?,
+    val durationBucket: AnalyticsDurationBucket?,
+    val appVersion: String?,
+    val osVersion: String?,
+    val payloadHash: String,
+) {
+    override fun toString(): String = "AnalyticsEvent([REDACTED])"
+}
+
+data class AnalyticsBatch(
+    val installationHash: String,
+    val accountId: UUID?,
+    val events: List<AnalyticsEvent>,
+    val receivedAt: Instant,
+) {
+    override fun toString(): String =
+        "AnalyticsBatch(installationHash=[REDACTED], accountId=[REDACTED], events=${events.size})"
+}
+
+data class AnalyticsIngestResult(
+    val accepted: Int,
+    val replayed: Int,
+)
+
+class AnalyticsEventTimeException :
+    ApiException(
+        status = HttpStatusCode.UnprocessableEntity,
+        code = "event_time_invalid",
+        message = "An event timestamp is outside the accepted range",
+    )

src/main/kotlin/com/osglab/account/features/analytics/models/AnalyticsDtos.kt

@@ -0,0 +1,48 @@
+package com.osglab.account.features.analytics.models
+
+import com.osglab.account.features.analytics.domain.AnalyticsAcquisitionChannel
+import com.osglab.account.features.analytics.domain.AnalyticsDurationBucket
+import com.osglab.account.features.analytics.domain.AnalyticsEventType
+import com.osglab.account.features.analytics.domain.AnalyticsExecutionMode
+import com.osglab.account.features.analytics.domain.AnalyticsFailureCategory
+import com.osglab.account.features.analytics.domain.AnalyticsFeature
+import com.osglab.account.features.analytics.domain.AnalyticsIngestResult
+import com.osglab.account.features.analytics.domain.AnalyticsSurface
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class AnalyticsBatchRequest(
+    val installationId: String,
+    val events: List<AnalyticsEventRequest>,
+) {
+    override fun toString(): String =
+        "AnalyticsBatchRequest(installationId=[REDACTED], events=[REDACTED size=${events.size}])"
+}
+
+@Serializable
+data class AnalyticsEventRequest(
+    val clientEventId: String,
+    val eventType: AnalyticsEventType,
+    val occurredAt: String,
+    val surface: AnalyticsSurface,
+    val acquisitionChannel: AnalyticsAcquisitionChannel? = null,
+    val feature: AnalyticsFeature? = null,
+    val executionMode: AnalyticsExecutionMode? = null,
+    val failureCategory: AnalyticsFailureCategory? = null,
+    val durationBucket: AnalyticsDurationBucket? = null,
+    val appVersion: String? = null,
+    val osVersion: String? = null,
+) {
+    override fun toString(): String = "AnalyticsEventRequest([REDACTED])"
+}
+
+@Serializable
+data class AnalyticsIngestResponse(
+    val accepted: Int,
+    val replayed: Int,
+) {
+    companion object {
+        fun fromDomain(result: AnalyticsIngestResult): AnalyticsIngestResponse =
+            AnalyticsIngestResponse(result.accepted, result.replayed)
+    }
+}

src/main/kotlin/com/osglab/account/features/analytics/repositories/AnalyticsRepository.kt

@@ -0,0 +1,197 @@
+package com.osglab.account.features.analytics.repositories
+
+import com.osglab.account.common.errors.ConflictException
+import com.osglab.account.config.DatabaseFactory
+import com.osglab.account.features.analytics.domain.AnalyticsAcquisitionChannel
+import com.osglab.account.features.analytics.domain.AnalyticsBatch
+import com.osglab.account.features.analytics.domain.AnalyticsDurationBucket
+import com.osglab.account.features.analytics.domain.AnalyticsEvent
+import com.osglab.account.features.analytics.domain.AnalyticsEventType
+import com.osglab.account.features.analytics.domain.AnalyticsExecutionMode
+import com.osglab.account.features.analytics.domain.AnalyticsFailureCategory
+import com.osglab.account.features.analytics.domain.AnalyticsFeature
+import com.osglab.account.features.analytics.domain.AnalyticsIngestResult
+import com.osglab.account.features.analytics.domain.AnalyticsSurface
+import org.jetbrains.exposed.v1.core.Table
+import org.jetbrains.exposed.v1.core.and
+import org.jetbrains.exposed.v1.core.eq
+import org.jetbrains.exposed.v1.core.inList
+import org.jetbrains.exposed.v1.core.isNull
+import org.jetbrains.exposed.v1.core.less
+import org.jetbrains.exposed.v1.core.plus
+import org.jetbrains.exposed.v1.javatime.date
+import org.jetbrains.exposed.v1.javatime.timestamp
+import org.jetbrains.exposed.v1.jdbc.insert
+import org.jetbrains.exposed.v1.jdbc.insertIgnore
+import org.jetbrains.exposed.v1.jdbc.deleteWhere
+import org.jetbrains.exposed.v1.jdbc.selectAll
+import org.jetbrains.exposed.v1.jdbc.update
+import java.time.Instant
+import java.time.ZoneOffset
+
+interface AnalyticsRepository {
+    suspend fun ingest(batch: AnalyticsBatch): AnalyticsIngestResult
+    suspend fun recordInvitePageOpen(occurredAt: Instant)
+    suspend fun purgeAnonymousInstallations(before: Instant, limit: Int): Int
+}
+
+class ExposedAnalyticsRepository(
+    private val databaseFactory: DatabaseFactory,
+) : AnalyticsRepository {
+    override suspend fun purgeAnonymousInstallations(before: Instant, limit: Int): Int {
+        require(limit in 1..10_000)
+        return databaseFactory.query {
+            val hashes = AnalyticsInstallations
+                .selectAll()
+                .where {
+                    AnalyticsInstallations.accountId.isNull() and
+                        (AnalyticsInstallations.updatedAt less before)
+                }
+                .limit(limit)
+                .map { it[AnalyticsInstallations.installationHash] }
+            if (hashes.isEmpty()) {
+                0
+            } else {
+                AnalyticsInstallations.deleteWhere {
+                    AnalyticsInstallations.installationHash inList hashes
+                }
+            }
+        }
+    }
+
+    override suspend fun recordInvitePageOpen(occurredAt: Instant) {
+        databaseFactory.query {
+            val date = occurredAt.atZone(ZoneOffset.UTC).toLocalDate()
+            AnalyticsDailyCounters.insertIgnore {
+                it[counterDate] = date
+                it[counterName] = INVITE_PAGE_OPENED
+                it[counterValue] = 0
+                it[updatedAt] = occurredAt
+            }
+            AnalyticsDailyCounters.update({
+                (AnalyticsDailyCounters.counterDate eq date) and
+                    (AnalyticsDailyCounters.counterName eq INVITE_PAGE_OPENED)
+            }) {
+                it[counterValue] = AnalyticsDailyCounters.counterValue + 1
+                it[updatedAt] = occurredAt
+            }
+        }
+    }
+
+    override suspend fun ingest(batch: AnalyticsBatch): AnalyticsIngestResult =
+        databaseFactory.query {
+            AnalyticsInstallations.insertIgnore {
+                it[installationHash] = batch.installationHash
+                it[accountId] = batch.accountId?.toString()
+                it[createdAt] = batch.receivedAt
+                it[updatedAt] = batch.receivedAt
+            }
+
+            val installation = AnalyticsInstallations
+                .selectAll()
+                .where { AnalyticsInstallations.installationHash eq batch.installationHash }
+                .forUpdate()
+                .single()
+            val linkedAccount = installation[AnalyticsInstallations.accountId]
+            when {
+                batch.accountId == null -> Unit
+                linkedAccount == null -> AnalyticsInstallations.update({
+                    AnalyticsInstallations.installationHash eq batch.installationHash
+                }) {
+                    it[accountId] = batch.accountId.toString()
+                    it[updatedAt] = batch.receivedAt
+                }
+                linkedAccount != batch.accountId.toString() ->
+                    throw ConflictException("Installation is linked to another account")
+            }
+            AnalyticsInstallations.update({
+                AnalyticsInstallations.installationHash eq batch.installationHash
+            }) {
+                it[updatedAt] = batch.receivedAt
+            }
+
+            var accepted = 0
+            var replayed = 0
+            batch.events.forEach { event ->
+                val existingPayloadHash = AnalyticsEvents
+                    .selectAll()
+                    .where {
+                        (AnalyticsEvents.installationHash eq batch.installationHash) and
+                            (AnalyticsEvents.clientEventId eq event.clientEventId.toString())
+                    }
+                    .singleOrNull()
+                    ?.get(AnalyticsEvents.payloadHash)
+                when {
+                    existingPayloadHash == null -> {
+                        insertEvent(batch, event)
+                        accepted += 1
+                    }
+                    existingPayloadHash == event.payloadHash -> replayed += 1
+                    else -> throw ConflictException("Client event ID was reused with another payload")
+                }
+            }
+            AnalyticsIngestResult(accepted = accepted, replayed = replayed)
+        }
+
+    private fun insertEvent(batch: AnalyticsBatch, event: AnalyticsEvent) {
+        AnalyticsEvents.insert {
+            it[installationHash] = batch.installationHash
+            it[clientEventId] = event.clientEventId.toString()
+            it[eventType] = event.eventType
+            it[occurredAt] = event.occurredAt
+            it[surface] = event.surface
+            it[acquisitionChannel] = event.acquisitionChannel
+            it[feature] = event.feature
+            it[executionMode] = event.executionMode
+            it[failureCategory] = event.failureCategory
+            it[durationBucket] = event.durationBucket
+            it[appVersion] = event.appVersion
+            it[osVersion] = event.osVersion
+            it[payloadHash] = event.payloadHash
+            it[receivedAt] = batch.receivedAt
+        }
+    }
+}
+
+private object AnalyticsInstallations : Table("product_analytics_installations") {
+    val installationHash = char("installation_hash", 64)
+    val accountId = varchar("account_id", 36).nullable()
+    val createdAt = timestamp("created_at")
+    val updatedAt = timestamp("updated_at")
+
+    override val primaryKey = PrimaryKey(installationHash)
+}
+
+private object AnalyticsEvents : Table("product_analytics_events") {
+    val installationHash = char("installation_hash", 64)
+    val clientEventId = char("client_event_id", 36)
+    val eventType = enumerationByName<AnalyticsEventType>("event_name", 32)
+    val occurredAt = timestamp("occurred_at")
+    val surface = enumerationByName<AnalyticsSurface>("surface", 16)
+    val acquisitionChannel =
+        enumerationByName<AnalyticsAcquisitionChannel>("acquisition_channel", 32).nullable()
+    val feature = enumerationByName<AnalyticsFeature>("feature", 32).nullable()
+    val executionMode =
+        enumerationByName<AnalyticsExecutionMode>("execution_mode", 16).nullable()
+    val failureCategory =
+        enumerationByName<AnalyticsFailureCategory>("failure_category", 32).nullable()
+    val durationBucket =
+        enumerationByName<AnalyticsDurationBucket>("duration_bucket", 16).nullable()
+    val appVersion = varchar("app_version", 32).nullable()
+    val osVersion = varchar("os_version", 32).nullable()
+    val payloadHash = char("payload_hash", 64)
+    val receivedAt = timestamp("received_at")
+
+    override val primaryKey = PrimaryKey(installationHash, clientEventId)
+}
+
+private object AnalyticsDailyCounters : Table("product_analytics_daily_counters") {
+    val counterDate = date("counter_date")
+    val counterName = varchar("counter_name", 32)
+    val counterValue = long("counter_value")
+    val updatedAt = timestamp("updated_at")
+
+    override val primaryKey = PrimaryKey(counterDate, counterName)
+}
+
+private const val INVITE_PAGE_OPENED = "INVITE_PAGE_OPENED"

src/main/kotlin/com/osglab/account/features/analytics/routes/AnalyticsRoutes.kt

@@ -0,0 +1,50 @@
+package com.osglab.account.features.analytics.routes
+
+import com.osglab.account.common.security.AccountPrincipal
+import com.osglab.account.common.security.SESSION_AUTH_NAME
+import com.osglab.account.common.errors.InvalidRequestException
+import com.osglab.account.features.analytics.models.AnalyticsBatchRequest
+import com.osglab.account.features.analytics.models.AnalyticsIngestResponse
+import com.osglab.account.features.analytics.services.AnalyticsService
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.HttpHeaders
+import io.ktor.server.auth.authenticate
+import io.ktor.server.auth.principal
+import io.ktor.server.request.header
+import io.ktor.server.request.receiveText
+import io.ktor.server.response.respond
+import io.ktor.server.routing.Route
+import io.ktor.server.routing.post
+import kotlinx.serialization.SerializationException
+import kotlinx.serialization.json.Json
+
+fun Route.analyticsRoutes(service: AnalyticsService) {
+    authenticate(SESSION_AUTH_NAME, optional = true) {
+        post("/v1/analytics/events") {
+            val declaredLength = call.request.header(HttpHeaders.ContentLength)?.toLongOrNull()
+            if (declaredLength != null && declaredLength > MAX_ANALYTICS_BODY_BYTES) {
+                throw InvalidRequestException("Analytics request body is too large")
+            }
+            val body = call.receiveText()
+            if (body.toByteArray(Charsets.UTF_8).size > MAX_ANALYTICS_BODY_BYTES) {
+                throw InvalidRequestException("Analytics request body is too large")
+            }
+            val request = try {
+                ANALYTICS_JSON.decodeFromString<AnalyticsBatchRequest>(body)
+            } catch (_: SerializationException) {
+                throw InvalidRequestException("Analytics request body is invalid")
+            }
+            val result = service.ingest(
+                accountId = call.principal<AccountPrincipal>()?.userId,
+                request = request,
+            )
+            call.respond(HttpStatusCode.OK, AnalyticsIngestResponse.fromDomain(result))
+        }
+    }
+}
+
+private val ANALYTICS_JSON = Json {
+    ignoreUnknownKeys = false
+    explicitNulls = false
+}
+private const val MAX_ANALYTICS_BODY_BYTES = 64 * 1024

src/main/kotlin/com/osglab/account/features/analytics/services/AnalyticsService.kt

@@ -0,0 +1,247 @@
+package com.osglab.account.features.analytics.services
+
+import com.osglab.account.common.errors.InvalidRequestException
+import com.osglab.account.features.analytics.domain.AnalyticsAcquisitionChannel
+import com.osglab.account.features.analytics.domain.AnalyticsBatch
+import com.osglab.account.features.analytics.domain.AnalyticsEvent
+import com.osglab.account.features.analytics.domain.AnalyticsEventTimeException
+import com.osglab.account.features.analytics.domain.AnalyticsEventType
+import com.osglab.account.features.analytics.domain.AnalyticsFailureCategory
+import com.osglab.account.features.analytics.domain.AnalyticsIngestResult
+import com.osglab.account.features.analytics.domain.AnalyticsSurface
+import com.osglab.account.features.analytics.models.AnalyticsBatchRequest
+import com.osglab.account.features.analytics.models.AnalyticsEventRequest
+import com.osglab.account.features.analytics.repositories.AnalyticsRepository
+import java.security.MessageDigest
+import java.time.Clock
+import java.time.Duration
+import java.time.Instant
+import java.time.format.DateTimeParseException
+import java.util.UUID
+
+interface AnalyticsService {
+    suspend fun ingest(
+        accountId: UUID?,
+        request: AnalyticsBatchRequest,
+    ): AnalyticsIngestResult
+}
+
+class DefaultAnalyticsService(
+    private val repository: AnalyticsRepository,
+    private val clock: Clock = Clock.systemUTC(),
+) : AnalyticsService {
+    override suspend fun ingest(
+        accountId: UUID?,
+        request: AnalyticsBatchRequest,
+    ): AnalyticsIngestResult {
+        if (request.events.size !in MIN_BATCH_SIZE..MAX_BATCH_SIZE) {
+            throw InvalidRequestException("events must contain between 1 and 50 items")
+        }
+        val installationId = parseUuid(request.installationId, "installationId")
+        val now = clock.instant()
+        val events = request.events.map { validateAndMap(it, now) }
+        return repository.ingest(
+            AnalyticsBatch(
+                installationHash = installationId.toString().sha256Hex(),
+                accountId = accountId,
+                events = events,
+                receivedAt = now,
+            )
+        )
+    }
+
+    private fun validateAndMap(request: AnalyticsEventRequest, now: Instant): AnalyticsEvent {
+        val clientEventId = parseUuid(request.clientEventId, "clientEventId")
+        val occurredAt = parseOccurredAt(request.occurredAt)
+        if (
+            occurredAt.isBefore(now.minus(MAX_EVENT_AGE)) ||
+            occurredAt.isAfter(now.plus(MAX_FUTURE_SKEW))
+        ) {
+            throw AnalyticsEventTimeException()
+        }
+        validateReleaseIdentifier(request.appVersion, "appVersion")
+        validateReleaseIdentifier(request.osVersion, "osVersion")
+        validateEventShape(request)
+
+        return AnalyticsEvent(
+            clientEventId = clientEventId,
+            eventType = request.eventType,
+            occurredAt = occurredAt,
+            surface = request.surface,
+            acquisitionChannel = request.acquisitionChannel,
+            feature = request.feature,
+            executionMode = request.executionMode,
+            failureCategory = request.failureCategory,
+            durationBucket = request.durationBucket,
+            appVersion = request.appVersion,
+            osVersion = request.osVersion,
+            payloadHash = payloadHash(request, clientEventId, occurredAt),
+        )
+    }
+
+    private fun validateEventShape(event: AnalyticsEventRequest) {
+        val valid = when (event.eventType) {
+            AnalyticsEventType.FIRST_OPEN ->
+                event.surface == AnalyticsSurface.APP &&
+                    event.acquisitionChannel != null &&
+                    event.feature == null &&
+                    event.executionMode == null &&
+                    event.failureCategory == null &&
+                    event.durationBucket == null
+
+            AnalyticsEventType.SESSION_STARTED ->
+                event.surface in setOf(AnalyticsSurface.APP, AnalyticsSurface.KEYBOARD) &&
+                    event.acquisitionChannel == null &&
+                    event.feature == null &&
+                    event.executionMode == null &&
+                    event.failureCategory == null &&
+                    event.durationBucket == null
+
+            AnalyticsEventType.KEYBOARD_ACTIVATED ->
+                event.surface == AnalyticsSurface.KEYBOARD &&
+                    event.acquisitionChannel == null &&
+                    event.feature == null &&
+                    event.executionMode == null &&
+                    event.failureCategory == null &&
+                    event.durationBucket == null
+
+            AnalyticsEventType.AI_FEATURE_STARTED ->
+                event.feature != null &&
+                    event.executionMode != null &&
+                    event.acquisitionChannel == null &&
+                    event.failureCategory == null &&
+                    event.durationBucket == null
+
+            AnalyticsEventType.AI_FEATURE_SUCCEEDED ->
+                event.feature != null &&
+                    event.executionMode != null &&
+                    event.durationBucket != null &&
+                    event.acquisitionChannel == null &&
+                    event.failureCategory == null
+
+            AnalyticsEventType.AI_FEATURE_FAILED ->
+                event.feature != null &&
+                    event.executionMode != null &&
+                    event.failureCategory != null &&
+                    event.acquisitionChannel == null
+
+            AnalyticsEventType.PURCHASE_VIEWED,
+            AnalyticsEventType.PURCHASE_STARTED,
+            ->
+                event.surface == AnalyticsSurface.APP &&
+                    event.acquisitionChannel == null &&
+                    event.feature == null &&
+                    event.executionMode == null &&
+                    event.failureCategory == null &&
+                    event.durationBucket == null
+
+            AnalyticsEventType.PURCHASE_CANCELLED ->
+                event.surface == AnalyticsSurface.APP &&
+                    event.failureCategory == AnalyticsFailureCategory.CANCELLED &&
+                    event.acquisitionChannel == null &&
+                    event.feature == null &&
+                    event.executionMode == null &&
+                    event.durationBucket == null
+
+            AnalyticsEventType.REFERRAL_SHARED ->
+                event.surface == AnalyticsSurface.APP &&
+                    event.acquisitionChannel == null &&
+                    event.feature == null &&
+                    event.executionMode == null &&
+                    event.failureCategory == null &&
+                    event.durationBucket == null
+
+            AnalyticsEventType.INVITE_OPENED ->
+                event.surface == AnalyticsSurface.INVITE_WEB &&
+                    event.acquisitionChannel == AnalyticsAcquisitionChannel.REFERRAL &&
+                    event.feature == null &&
+                    event.executionMode == null &&
+                    event.failureCategory == null &&
+                    event.durationBucket == null
+        }
+        if (!valid) {
+            throw InvalidRequestException("Event fields do not match the event type")
+        }
+    }
+
+    private fun parseOccurredAt(value: String): Instant {
+        if (!value.endsWith('Z')) {
+            throw InvalidRequestException("occurredAt must be a UTC ISO-8601 timestamp")
+        }
+        return try {
+            Instant.parse(value)
+        } catch (_: DateTimeParseException) {
+            throw InvalidRequestException("occurredAt must be a UTC ISO-8601 timestamp")
+        }
+    }
+
+    private fun validateReleaseIdentifier(value: String?, field: String) {
+        if (value != null && !RELEASE_IDENTIFIER.matches(value)) {
+            throw InvalidRequestException("$field must be a 1 to 32 character release identifier")
+        }
+    }
+
+    private fun parseUuid(value: String, field: String): UUID {
+        if (!UUID_PATTERN.matches(value)) {
+            throw InvalidRequestException("$field must be a UUID")
+        }
+        return try {
+            UUID.fromString(value)
+        } catch (_: IllegalArgumentException) {
+            throw InvalidRequestException("$field must be a UUID")
+        }
+    }
+
+    private fun payloadHash(
+        event: AnalyticsEventRequest,
+        clientEventId: UUID,
+        occurredAt: Instant,
+    ): String = listOf(
+        clientEventId.toString(),
+        event.eventType.name,
+        occurredAt.toString(),
+        event.surface.name,
+        event.acquisitionChannel?.name,
+        event.feature?.name,
+        event.executionMode?.name,
+        event.failureCategory?.name,
+        event.durationBucket?.name,
+        event.appVersion,
+        event.osVersion,
+    ).joinToString(separator = "\u0000") { it ?: "" }.sha256Hex()
+
+    private fun String.sha256Hex(): String =
+        MessageDigest.getInstance("SHA-256")
+            .digest(toByteArray(Charsets.UTF_8))
+            .joinToString(separator = "") { byte -> "%02x".format(byte.toInt() and 0xff) }
+
+    private companion object {
+        const val MIN_BATCH_SIZE = 1
+        const val MAX_BATCH_SIZE = 50
+        val MAX_EVENT_AGE: Duration = Duration.ofDays(35)
+        val MAX_FUTURE_SKEW: Duration = Duration.ofMinutes(5)
+        val UUID_PATTERN =
+            Regex("[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
+        val RELEASE_IDENTIFIER = Regex("[A-Za-z0-9._+\\-]{1,32}")
+    }
+}
+
+class AnalyticsMaintenanceService(
+    private val repository: AnalyticsRepository,
+    private val clock: Clock = Clock.systemUTC(),
+    private val anonymousRetention: Duration = Duration.ofDays(90),
+) {
+    init {
+        require(!anonymousRetention.isNegative && !anonymousRetention.isZero)
+    }
+
+    suspend fun purgeStaleAnonymousInstallations(): Int =
+        repository.purgeAnonymousInstallations(
+            before = clock.instant().minus(anonymousRetention),
+            limit = PURGE_BATCH_SIZE,
+        )
+
+    private companion object {
+        const val PURGE_BATCH_SIZE = 1_000
+    }
+}

src/main/kotlin/com/osglab/account/features/inviteweb/InviteWebRoutes.kt

@@ -27,6 +27,10 @@ fun interface ReferralLookupPort {
     suspend fun isValid(code: String): Boolean
 }
 
+fun interface InviteOpenRecorder {
+    suspend fun record()
+}
+
 data class InviteWebConfig(
     val appStoreUrl: String,
     val appleAppId: String,
@@ -75,6 +79,7 @@ sealed interface InvitePageResult {
 class InvitePageService(
     private val referralLookup: ReferralLookupPort,
     private val config: InviteWebConfig,
+    private val inviteOpenRecorder: InviteOpenRecorder = InviteOpenRecorder {},
 ) {
     private val appStoreUrl = validateHttpsUrl(config.appStoreUrl, "APP_STORE_URL", allowQuery = true)
         .toASCIIString()
@@ -101,6 +106,13 @@ class InvitePageService(
             return InvitePageResult.TemporarilyUnavailable
         }
         if (!valid) return InvitePageResult.Invalid
+        try {
+            inviteOpenRecorder.record()
+        } catch (exception: CancellationException) {
+            throw exception
+        } catch (_: Exception) {
+            // Analytics must never make a valid invitation unavailable.
+        }
 
         val nonce = createNonce()
         val universalLink = "$universalLinkBaseUrl/$validCode".escapeHtml()
@@ -129,6 +141,7 @@ fun Route.configureInviteWebRoutes() {
     }
     configureInviteWebRoutes(
         referralLookup = koin.get<ReferralLookupPort>(),
+        inviteOpenRecorder = koin.get<InviteOpenRecorder>(),
         config = InviteWebConfig(
             appStoreUrl = appConfig.appStoreUrl,
             appleAppId = "$teamId.${appConfig.apple.clientId}",
@@ -140,8 +153,9 @@ fun Route.configureInviteWebRoutes() {
 fun Route.configureInviteWebRoutes(
     referralLookup: ReferralLookupPort,
     config: InviteWebConfig,
+    inviteOpenRecorder: InviteOpenRecorder = InviteOpenRecorder {},
 ) {
-    val service = InvitePageService(referralLookup, config)
+    val service = InvitePageService(referralLookup, config, inviteOpenRecorder)
 
     get("/i/{code}") {
         when (val result = service.render(call.parameters["code"])) {