import { and, count, countDistinct, eq, inArray, isNotNull, lt, ne, notInArray, sql } from 'drizzle-orm'

import { db, schema, type DbOrTx } from '../db'
import { publishAchievementNotification } from '../events'
import { calendarDay, dayNumber, localHour, localTime, weekStart } from '../time'
import { findMetric } from './achievement-metrics'
import { isAccepted, JudgeStatus } from '../judge/status'
import { asRecord } from '../routes/helpers'

function numberMetric(metrics: Record<string, unknown>, key: string) {
  const value = metrics[key]
  return typeof value === 'number' ? value : 0
}

/**
 * 「连续几周都有新通过」：`week` 是这次新通过所在那周的周一（东八区日历日）。
 * 和上一次新通过同一周不变；正好下一周就接上；隔了一周以上从 1 重来。
 *
 * 比「坚持三天」那段多一道判断：这次的周比记下的还早（重判了一条很早以前的提交）
 * 就什么都不动 —— 否则会把「当前连续」错误地重置成 1、还把 `_last_ac_week` 拨回过去。
 */
function nextWeekStreak(value: Record<string, unknown>, week: string) {
  const last = typeof value._last_ac_week === 'string' ? value._last_ac_week : null
  if (last && week <= last) return {}
  const current = last && dayNumber(week) - dayNumber(last) === 7 ? numberMetric(value, '_current_ac_week_streak') + 1 : 1
  return {
    _last_ac_week: week,
    _current_ac_week_streak: current,
    max_ac_week_streak: Math.max(numberMetric(value, 'max_ac_week_streak'), current),
  }
}

async function unlockAchievements(userId: number, metrics: Record<string, unknown>, onlyMeta = false) {
  const unlocked = await db.select({ id: schema.userAchievement.achievementId }).from(schema.userAchievement).where(eq(schema.userAchievement.userId, userId))
  const filters = [eq(schema.achievement.visible, true)]
  if (unlocked.length)
    filters.push(
      notInArray(
        schema.achievement.id,
        unlocked.map((row) => row.id),
      ),
    )
  if (onlyMeta) filters.push(eq(schema.achievement.metric, 'achievement_unlocked_count'))
  else filters.push(ne(schema.achievement.metric, 'achievement_unlocked_count'))
  const candidates = await db
    .select()
    .from(schema.achievement)
    .where(and(...filters))
  const hits = candidates.filter((achievement) => {
    const value = metrics[achievement.metric]
    if (typeof value !== 'number') return false
    return achievement.operator === 'gte' ? value >= achievement.threshold : value <= achievement.threshold
  })
  if (hits.length === 0) return []
  // 命中的成就一次插完，冲突忽略后 returning 回来的就是「这次真新解锁的」。
  // 一个用户对同一个成就只会解锁一次，所以每个成就都恰好 +1，一条 UPDATE 就够。
  const inserted = await db
    .insert(schema.userAchievement)
    .values(
      hits.map((achievement) => ({
        userId,
        achievementId: achievement.id,
        unlockTime: new Date().toISOString(),
        backfilled: false,
        notified: false,
      })),
    )
    .onConflictDoNothing({
      target: [schema.userAchievement.achievementId, schema.userAchievement.userId],
    })
    .returning({ achievementId: schema.userAchievement.achievementId })
  if (inserted.length === 0) return []
  const insertedIds = new Set(inserted.map((row) => row.achievementId))
  await db
    .update(schema.achievement)
    .set({ unlockCount: sql`${schema.achievement.unlockCount} + 1` })
    .where(inArray(schema.achievement.id, [...insertedIds]))
  return hits.filter((achievement) => insertedIds.has(achievement.id))
}

const ACCEPTED_RESULTS = sql.join([sql`${JudgeStatus.ACCEPTED}`, sql`${JudgeStatus.AST_CHECK_FAILED}`], sql`, `)

/**
 * 「做对几道 / 中等几道 / 困难几道 / 一次就对几道」按定义从提交表现数，不再逐次 +1。
 * 参加过几场比赛、挑战成功几次也在这里一起数。
 *
 * 原来是读一遍这道题之前的提交、判断「是不是第一次做对」再 +1，两处漏：同一个人同一道题
 * 两条提交挨着判完，两边都看见对方已经做对（都不加）或都还没做对（加两次）；重判一条
 * 做对的提交又会再加一次。2026-10 的库里 372 人的做对题数比实际多 1–4 道。
 * 现数是幂等的，怎么并发、怎么重判都是同一个答案。走 submission_public_metrics_idx，
 * 一个人的提交也就几百条。
 *
 * **比赛里的提交也算**（2026-11-03 起）：这里的比赛其实是单班测验，有的班整学期都在
 * 比赛里做题，不算的话一个成就都拿不到。比赛题是单独的题目行，不会和题库里同一道题重复数。
 * 个人主页 / 排名的做对题数（user_profile.accepted_number）仍然不含比赛，两边对不上是有意的。
 */
async function solvedCounts(tx: DbOrTx, userId: number) {
  const [row] = await tx.execute<{
    accepted: number
    mid: number
    hard: number
    first_try: number
    contests: number
    challenges: number
  }>(
    sql`
      with mine as (
        select s.problem_id, s.contest_id, s.result, s.create_time, s.id, p.difficulty
        from ${schema.submission} s
        join ${schema.problem} p on p.id = s.problem_id
        where s.user_id = ${userId}
      )
      select
        count(distinct problem_id) filter (where result in (${ACCEPTED_RESULTS}))::int as accepted,
        count(distinct problem_id) filter (where result in (${ACCEPTED_RESULTS}) and difficulty = 'Mid')::int as mid,
        count(distinct problem_id) filter (where result in (${ACCEPTED_RESULTS}) and difficulty = 'High')::int as hard,
        (
          select count(*)::int from (
            select distinct on (problem_id) result from mine order by problem_id, create_time, id
          ) first where result in (${ACCEPTED_RESULTS})
        ) as first_try,
        count(distinct contest_id)::int as contests,
        (select count(*)::int from ${schema.challengeSuccess} where user_id = ${userId}) as challenges
      from mine
    `,
  )
  return row ?? { accepted: 0, mid: 0, hard: 0, first_try: 0, contests: 0, challenges: 0 }
}

/**
 * `countedAs`：这条提交上次结算时的结果，重判时由判题传进来；新提交是 null。
 * 重判不是又交了一次 —— 提交次数、凌晨/早起这类「每交一次 +1」的不能再加，
 * 编译错误按前后差值改。
 */
export async function updateAchievementsForSubmission(submissionId: string, countedAs: number | null = null) {
  const [row] = await db
    .select({ submission: schema.submission, problem: schema.problem })
    .from(schema.submission)
    .innerJoin(schema.problem, eq(schema.submission.problemId, schema.problem.id))
    .where(eq(schema.submission.id, submissionId))
    .limit(1)
  if (!row) return []

  const rejudged = countedAs !== null
  const accepted = isAccepted(row.submission.result)
  const date = calendarDay(row.submission.createTime)
  const hour = localHour(row.submission.createTime)

  const metrics = await db.transaction(async (tx) => {
    await tx
      .insert(schema.userStat)
      .values({
        userId: row.submission.userId,
        metrics: {},
        updateTime: new Date().toISOString(),
      })
      .onConflictDoNothing({ target: schema.userStat.userId })
    const [stat] = await tx.select().from(schema.userStat).where(eq(schema.userStat.userId, row.submission.userId)).for('update')
    if (!stat) throw new Error('User achievement stat could not be created')
    const value = asRecord(stat.metrics)
    // 在锁里数：同一个人的结算排成一队，后来的一定看得见先来的写进去的结果
    const solved = await solvedCounts(tx, row.submission.userId)
    // 这次让做对的题多了一道，才算「新做对」：每天做对几道、连续几周这些跟着它走
    const firstAc = accepted && solved.accepted > numberMetric(value, 'accepted_count')
    value.accepted_count = solved.accepted
    value.mid_ac_count = solved.mid
    value.hard_ac_count = solved.hard
    value.first_try_ac_count = solved.first_try
    value.contest_joined = solved.contests
    value.challenge_count = solved.challenges
    if (!rejudged) value.submission_count = numberMetric(value, 'submission_count') + 1
    if (firstAc) {
      // 只数这条之前的：重判一条老提交时，它后面的提交不算「做对之前错的」
      const [prior] = await tx
        .select({ value: count() })
        .from(schema.submission)
        .where(
          and(
            eq(schema.submission.userId, row.submission.userId),
            eq(schema.submission.problemId, row.submission.problemId),
            lt(schema.submission.createTime, row.submission.createTime),
          ),
        )
      value.max_wa_before_ac = Math.max(numberMetric(value, 'max_wa_before_ac'), prior?.value ?? 0)
      const perDay = asRecord(value._ac_per_day)
      perDay[date] = (typeof perDay[date] === 'number' ? perDay[date] : 0) + 1
      value._ac_per_day = perDay
      value.max_ac_in_one_day = Math.max(...Object.values(perDay).filter((item): item is number => typeof item === 'number'))
      Object.assign(value, nextWeekStreak(value, calendarDay(weekStart(row.submission.createTime))))
    }
    const activeDates = Array.isArray(value._active_dates) ? value._active_dates.filter((item): item is string => typeof item === 'string') : []
    if (!activeDates.includes(date)) activeDates.push(date)
    value._active_dates = activeDates
    value.active_days = activeDates.length
    if (accepted) {
      const last = typeof value._last_ac_date === 'string' ? value._last_ac_date : null
      // 比记下的还早（重判了一条以前的提交）就不动，和 nextWeekStreak 一样
      if (!last || date > last) {
        // 差一天要按日历日算，不能用 Date 相减：夏令时地区相邻两天差 23/25 小时，
        // 除 86400000 得到的不是 1，`=== 1` 会静默把连续打卡判成断掉。
        const current = last && dayNumber(date) - dayNumber(last) === 1 ? numberMetric(value, '_current_ac_streak') + 1 : 1
        value._last_ac_date = date
        value._current_ac_streak = current
        value.max_ac_streak_days = Math.max(numberMetric(value, 'max_ac_streak_days'), current)
      }
    }
    const languages = Array.isArray(value._languages) ? value._languages.filter((item): item is string => typeof item === 'string') : []
    if (!languages.includes(row.submission.language)) languages.push(row.submission.language)
    value._languages = languages
    value.languages_used = languages.length
    if (!rejudged) {
      if (hour < 5) value.midnight_submissions = numberMetric(value, 'midnight_submissions') + 1
      else if (hour < 7) value.early_bird_submissions = numberMetric(value, 'early_bird_submissions') + 1
    }
    // 编译错误是从上线那天起逐次累计的（之前的历史没补），没法按定义重数，只能按差值改
    const compileError = Number(row.submission.result === JudgeStatus.COMPILE_ERROR)
    const wasCompileError = Number(countedAs === JudgeStatus.COMPILE_ERROR)
    if (compileError !== wasCompileError) value.compile_error_count = Math.max(0, numberMetric(value, 'compile_error_count') + compileError - wasCompileError)
    // 只算做对的、空行不算：原来交什么都算，贴一大段就拿到「长篇巨制」（0032 按这个口径重算过存量）
    if (accepted)
      value.max_code_lines = Math.max(numberMetric(value, 'max_code_lines'), row.submission.code.split('\n').filter((line) => line.trim() !== '').length)
    await tx.update(schema.userStat).set({ metrics: value, updateTime: new Date().toISOString() }).where(eq(schema.userStat.id, stat.id))
    return value
  })

  const first = await unlockAchievements(row.submission.userId, metrics)
  if (!first.length) return []
  const [meta] = await db
    .select({ value: count() })
    .from(schema.userAchievement)
    .innerJoin(schema.achievement, eq(schema.userAchievement.achievementId, schema.achievement.id))
    .where(and(eq(schema.userAchievement.userId, row.submission.userId), ne(schema.achievement.rarity, 'platinum'), eq(schema.achievement.visible, true)))
  const unlockedCount = meta?.value ?? 0
  metrics.achievement_unlocked_count = unlockedCount
  await setUnlockedCount(row.submission.userId, unlockedCount)
  return [...first, ...(await unlockAchievements(row.submission.userId, metrics, true))]
}

/**
 * 只 `jsonb_set` 这一个键：这一步在锁外，整体写回 metrics 会把并发判题刚写的其它指标
 * 冲掉（和 refreshUnlockedCount 同一个理由）。
 */
async function setUnlockedCount(userId: number, value: number) {
  await db
    .update(schema.userStat)
    .set({
      metrics: sql`jsonb_set(${schema.userStat.metrics}, '{achievement_unlocked_count}', to_jsonb(${value}::int))`,
      updateTime: new Date().toISOString(),
    })
    .where(eq(schema.userStat.userId, userId))
}

export async function updateAchievementsForProblemSet(userId: number) {
  const [[badgeRow], [completedRow]] = await Promise.all([
    db.select({ value: count() }).from(schema.userBadge).where(eq(schema.userBadge.userId, userId)),
    db
      .select({ value: count() })
      .from(schema.problemsetProgress)
      .where(and(eq(schema.problemsetProgress.userId, userId), eq(schema.problemsetProgress.isCompleted, true))),
  ])
  const metrics = await db.transaction(async (tx) => {
    await tx
      .insert(schema.userStat)
      .values({
        userId,
        metrics: {},
        updateTime: new Date().toISOString(),
      })
      .onConflictDoNothing({ target: schema.userStat.userId })
    const [stat] = await tx.select().from(schema.userStat).where(eq(schema.userStat.userId, userId)).for('update').limit(1)
    if (!stat) throw new Error('User achievement stat could not be created')
    const value = asRecord(stat.metrics)
    value.badge_count = badgeRow?.value ?? 0
    value.problemset_completed = completedRow?.value ?? 0
    await tx.update(schema.userStat).set({ metrics: value, updateTime: new Date().toISOString() }).where(eq(schema.userStat.id, stat.id))
    return value
  })

  const first = await unlockAchievements(userId, metrics)
  if (!first.length) return []
  const [meta] = await db
    .select({ value: count() })
    .from(schema.userAchievement)
    .innerJoin(schema.achievement, eq(schema.userAchievement.achievementId, schema.achievement.id))
    .where(and(eq(schema.userAchievement.userId, userId), ne(schema.achievement.rarity, 'platinum'), eq(schema.achievement.visible, true)))
  const unlockedCount = meta?.value ?? 0
  metrics.achievement_unlocked_count = unlockedCount
  await setUnlockedCount(userId, unlockedCount)
  return [...first, ...(await unlockAchievements(userId, metrics, true))]
}

/** 一条 INSERT 里塞多少行。5 个参数一行，离 Postgres 的 65535 个参数上限还很远 */
const USER_ACHIEVEMENT_INSERT_CHUNK = 1000

/**
 * 新建成就、调低阈值、或从下架改成上架之后，把已达标的存量用户补发一遍。
 *
 * 判定平时只在判题结算时发生，后台改了配置不会自动补发，必须显式扫。
 * 对齐旧 `rescan_achievement`：只处理 visible 的成就，逐个用户 unlock，
 * 且标记 backfilled=true —— 这是补发不是刚挣到的，前端据此只显示「已获得」
 * 而不显示日期，否则一次补发会给几百人盖同一个时间戳，把「最近获得」板块冲垮。
 */
export async function rescanAchievement(achievementId: number) {
  const [achievement] = await db
    .select()
    .from(schema.achievement)
    .where(and(eq(schema.achievement.id, achievementId), eq(schema.achievement.visible, true)))
    .limit(1)
  if (!achievement) return { scanned: 0, unlocked: 0 }

  const metric = findMetric(achievement.metric)
  if (!metric) return { scanned: 0, unlocked: 0 }

  // contest_joined 判题结算从 2026-11-03 起才维护，那之后没再交过题的人还是旧值，扫之前刷新一遍
  if (achievement.metric === 'contest_joined') await refreshContestJoinedForAll()
  // max_ac_week_streak 是后加的指标，判题结算只从上线那天起累计，存量用户要从提交记录补算
  if (achievement.metric === 'max_ac_week_streak') await refreshAcWeekStreakForAll()

  const already = new Set(
    (
      await db.select({ userId: schema.userAchievement.userId }).from(schema.userAchievement).where(eq(schema.userAchievement.achievementId, achievement.id))
    ).map((row) => row.userId),
  )

  const stats = await db
    .select({
      userId: schema.userStat.userId,
      metrics: schema.userStat.metrics,
    })
    .from(schema.userStat)
  const eligible = stats.filter((stat) => {
    if (already.has(stat.userId)) return false
    const value = asRecord(stat.metrics)[achievement.metric]
    if (typeof value !== 'number') return false
    return achievement.operator === 'gte' ? value >= achievement.threshold : value <= achievement.threshold
  })

  // 达标的人分批插，冲突忽略后 returning 回来的就是真新解锁的那批 —— 以前是每人
  // 一条 insert 加一条 unlockCount+1，全站补发一次就是几千次往返。
  // 计数改成一次 +N，通知照旧逐人推（那是 Redis，不是数据库）。
  const unlockTime = new Date().toISOString()
  const unlockedUserIds: number[] = []
  for (let start = 0; start < eligible.length; start += USER_ACHIEVEMENT_INSERT_CHUNK) {
    const chunk = eligible.slice(start, start + USER_ACHIEVEMENT_INSERT_CHUNK)
    const inserted = await db
      .insert(schema.userAchievement)
      .values(
        chunk.map((stat) => ({
          userId: stat.userId,
          achievementId: achievement.id,
          unlockTime,
          backfilled: true,
          notified: false,
        })),
      )
      .onConflictDoNothing({
        target: [schema.userAchievement.achievementId, schema.userAchievement.userId],
      })
      .returning({ userId: schema.userAchievement.userId })
    unlockedUserIds.push(...inserted.map((row) => row.userId))
  }
  if (unlockedUserIds.length) {
    await db
      .update(schema.achievement)
      .set({
        unlockCount: sql`${schema.achievement.unlockCount} + ${unlockedUserIds.length}`,
      })
      .where(eq(schema.achievement.id, achievement.id))
    for (const userId of unlockedUserIds) {
      await publishAchievementNotification(userId, [
        {
          id: achievement.id,
          name: achievement.name,
          description: achievement.description,
          icon: achievement.icon,
          rarity: achievement.rarity,
          kind: 'achievement',
        },
      ])
    }
    // 补发的非白金成就同样计入「已解锁数」，要和判题结算一样接着做第二轮（元成就）判定。
    // 旧 `rescan_achievement` 就漏了这步，OJ2 原样搬过来：2026-09-07 一次补发之后
    // 269 人的计数停在旧值，其中 10 人实际够了「奖杯收藏家」却一直没发 ——
    // 判题结算只在「这次有新解锁」时才重算，被补发的人不再解锁新成就就永远不会自愈。
    if (achievement.rarity !== 'platinum' && achievement.metric !== 'achievement_unlocked_count') {
      await refreshUnlockedCount(unlockedUserIds)
      for (const meta of await metaAchievements()) await rescanAchievement(meta.id)
    }
  }
  return { scanned: stats.length, unlocked: unlockedUserIds.length }
}

/** 以「已解锁数」为指标的元成就（奖杯收藏家）。只取上架的，和 rescan 的口径一致 */
export function metaAchievements() {
  return db
    .select({
      id: schema.achievement.id,
      name: schema.achievement.name,
      threshold: schema.achievement.threshold,
      operator: schema.achievement.operator,
    })
    .from(schema.achievement)
    .where(and(eq(schema.achievement.visible, true), eq(schema.achievement.metric, 'achievement_unlocked_count')))
}

/**
 * 按 `user_achievement` 重算 `achievement_unlocked_count`，返回实际改动了的用户 id。
 *
 * 口径和判题结算一致：已解锁的、还上架的**非白金**成就数（下架的不算，见 0032）。只 `jsonb_set` 这一个键、只写值变了的行，
 * 不整体覆盖 `metrics` —— 整体写回会和并发判题写的其它指标互相踩。
 * 不传 `userIds` 就是全体有 `user_stat` 的用户（`recount` 存量订正用）。
 */
export async function refreshUnlockedCount(userIds?: number[]) {
  if (userIds && userIds.length === 0) return []
  const scope = userIds ? sql`and s.user_id in ${userIds}` : sql``
  const rows = await db.execute<{ user_id: number }>(sql`
    update ${schema.userStat} as target
    set metrics = jsonb_set(target.metrics, '{achievement_unlocked_count}', to_jsonb(fresh.value))
    from (
      select s.id, coalesce(c.value, 0) as value
      from ${schema.userStat} s
      left join (
        select ua.user_id, count(*)::int as value
        from ${schema.userAchievement} ua
        join ${schema.achievement} a on a.id = ua.achievement_id
        where a.rarity <> 'platinum' and a.visible
        group by ua.user_id
      ) c on c.user_id = s.user_id
      where true ${scope}
    ) fresh
    where target.id = fresh.id
      and (target.metrics -> 'achievement_unlocked_count') is distinct from to_jsonb(fresh.value)
    returning target.user_id
  `)
  return rows.map((row) => row.user_id)
}

/** 同上，3 个参数一行 */
const STAT_UPSERT_CHUNK = 1000

/**
 * 把所有有过比赛提交的用户的 contest_joined 重算一遍，供 rescan 前置调用。
 *
 * 参赛场次这个指标：旧后端 `ContestJoined` 只实现了 recompute、不走 on_submission
 * （比赛提交在 build_ctx 就被跳过了），所以它只在 rescan 时刷新。这里保持同样口径：
 * 去重数一遍该用户有过提交的比赛数。
 * （迁移过来时新后端整个漏掉过这个指标，配在 contest_joined 上的成就永远解锁不了。）
 *
 * 一条 group by 出全部用户的场次，再分批 upsert —— 以前是每个用户 1 条 count
 * 加一个独立事务里的 insert/select for update/update，四次往返乘以全站用户数。
 *
 * `metrics || excluded.metrics` 是 jsonb 的浅合并，只覆盖 contest_joined 这一个键，
 * 其余指标原样保留，语义和原来「读出来改一个键再写回去」一致，而且不需要 for update：
 * 合并在一条语句里完成，并发写不会互相盖掉。
 */
async function refreshContestJoinedForAll() {
  const rows = await db
    .select({
      userId: schema.submission.userId,
      value: countDistinct(schema.submission.contestId),
    })
    .from(schema.submission)
    // 只算还在的用户，理由同 refreshAcWeekStreakForAll：库里有 11 个已不存在的用户
    // 留下了比赛提交，不关联的话这里 upsert 撞外键，比赛类成就的重扫一直是失败的
    .innerJoin(schema.user, eq(schema.user.id, schema.submission.userId))
    .where(isNotNull(schema.submission.contestId))
    .groupBy(schema.submission.userId)
  const now = new Date().toISOString()
  for (let start = 0; start < rows.length; start += STAT_UPSERT_CHUNK) {
    const chunk = rows.slice(start, start + STAT_UPSERT_CHUNK)
    await db
      .insert(schema.userStat)
      .values(
        chunk.map((row) => ({
          userId: row.userId,
          metrics: { contest_joined: row.value },
          updateTime: now,
        })),
      )
      .onConflictDoUpdate({
        target: schema.userStat.userId,
        set: {
          metrics: sql`${schema.userStat.metrics} || excluded.metrics`,
          updateTime: sql`excluded.update_time`,
        },
      })
  }
}

/**
 * 从提交记录重算所有人的「最长连续新通过周数」，连同接着累计要用的
 * `_last_ac_week` / `_current_ac_week_streak` 一起写回 user_stat（jsonb 浅合并，
 * 其余指标不动，写法同 refreshContestJoinedForAll）。
 *
 * 「新通过」= 这道题（含比赛题）第一次通过的那次，和判题结算里 firstAc 同一个口径；
 * 周按东八区自然周（周一起），SQL 里用 localTime 取墙上时间再 date_trunc。
 */
async function refreshAcWeekStreakForAll() {
  const firstAccepted = db
    .select({
      userId: schema.submission.userId,
      firstAt: sql<string>`min(${schema.submission.createTime})`.as('first_at'),
    })
    .from(schema.submission)
    // 只算还在的用户：库里有提交的 user_id 已经对不上任何用户（submission.user_id 上没有
    // 外键，是历史遗留），直接写 user_stat 会撞 user_stat 的外键，整次重算失败
    .innerJoin(schema.user, eq(schema.user.id, schema.submission.userId))
    .where(inArray(schema.submission.result, [JudgeStatus.ACCEPTED, JudgeStatus.AST_CHECK_FAILED]))
    .groupBy(schema.submission.userId, schema.submission.problemId)
    .as('first_accepted')
  const week = sql<string>`to_char(date_trunc('week', ${localTime(firstAccepted.firstAt)}), 'YYYY-MM-DD')`
  const rows = await db
    .select({ userId: firstAccepted.userId, week })
    .from(firstAccepted)
    .groupBy(firstAccepted.userId, week)
    .orderBy(firstAccepted.userId, week)

  const byUser = new Map<number, Record<string, unknown>>()
  for (const row of rows) {
    const value = byUser.get(row.userId) ?? {}
    Object.assign(value, nextWeekStreak(value, row.week))
    byUser.set(row.userId, value)
  }

  const entries = [...byUser]
  const now = new Date().toISOString()
  for (let start = 0; start < entries.length; start += STAT_UPSERT_CHUNK) {
    const chunk = entries.slice(start, start + STAT_UPSERT_CHUNK)
    await db
      .insert(schema.userStat)
      .values(chunk.map(([userId, metrics]) => ({ userId, metrics, updateTime: now })))
      .onConflictDoUpdate({
        target: schema.userStat.userId,
        set: {
          metrics: sql`${schema.userStat.metrics} || excluded.metrics`,
          updateTime: sql`excluded.update_time`,
        },
      })
  }
}