From 3ee609337c8433c697a92711fe59eddd4f0887c2 Mon Sep 17 00:00:00 2001 From: Amir Husayn Panahifar Date: Wed, 7 Oct 2026 19:40:26 +0330 Subject: [PATCH] feat(backend): add domain store, learning services, and seed data Persist LMS entities and implement hearts/watch rules plus curriculum bootstrap. --- backend/internal/seed/seed.go | 249 +++ backend/internal/service/hearts.go | 94 + backend/internal/service/hearts_staff_test.go | 28 + backend/internal/service/notify.go | 61 + backend/internal/service/util.go | 21 + backend/internal/service/watch.go | 112 ++ backend/internal/service/watch_test.go | 50 + backend/internal/store/blacklist.go | 281 +++ backend/internal/store/challenges.go | 135 ++ backend/internal/store/chat_export.go | 243 +++ backend/internal/store/invites.go | 450 +++++ backend/internal/store/learning.go | 314 +++ backend/internal/store/learning_test.go | 36 + backend/internal/store/store.go | 1764 +++++++++++++++++ backend/internal/store/whitelist.go | 238 +++ 15 files changed, 4076 insertions(+) create mode 100644 backend/internal/seed/seed.go create mode 100644 backend/internal/service/hearts.go create mode 100644 backend/internal/service/hearts_staff_test.go create mode 100644 backend/internal/service/notify.go create mode 100644 backend/internal/service/util.go create mode 100644 backend/internal/service/watch.go create mode 100644 backend/internal/service/watch_test.go create mode 100644 backend/internal/store/blacklist.go create mode 100644 backend/internal/store/challenges.go create mode 100644 backend/internal/store/chat_export.go create mode 100644 backend/internal/store/invites.go create mode 100644 backend/internal/store/learning.go create mode 100644 backend/internal/store/learning_test.go create mode 100644 backend/internal/store/store.go create mode 100644 backend/internal/store/whitelist.go diff --git a/backend/internal/seed/seed.go b/backend/internal/seed/seed.go new file mode 100644 index 0000000..0cce526 --- /dev/null +++ b/backend/internal/seed/seed.go @@ -0,0 +1,249 @@ +package seed + +import ( + "context" + "fmt" + "log" + "os" + "strings" + "time" + + "golang.org/x/crypto/bcrypt" + + "pargar/backend/internal/models" + "pargar/backend/internal/store" +) + +var demoEmails = []string{ + "student@pargar.dev", + "mentor@pargar.dev", + "admin@pargar.dev", +} + +// Run ensures the bootstrap admin exists, removes legacy demo accounts, +// and optionally seeds an idempotent curriculum when the database is empty. +func Run(ctx context.Context, st *store.Store, allowCurriculumSeed bool) error { + if err := purgeDemoUsers(ctx, st); err != nil { + return err + } + if err := ensureBootstrapAdmin(ctx, st); err != nil { + return err + } + + if !allowCurriculumSeed { + return nil + } + + chapters, _ := st.ListChapters(ctx) + if len(chapters) > 0 { + return nil + } + + if err := seedCurriculum(ctx, st); err != nil { + return err + } + return seedEvents(ctx, st) +} + +func purgeDemoUsers(ctx context.Context, st *store.Store) error { + for _, email := range demoEmails { + u, err := st.GetUserByEmail(ctx, email) + if err != nil { + continue + } + if err := st.DeleteUser(ctx, u.ID); err != nil { + return err + } + log.Printf("removed legacy demo user %s", email) + } + return nil +} + +func ensureBootstrapAdmin(ctx context.Context, st *store.Store) error { + name := envOr("ADMIN_NAME", "Admin") + username := strings.ToLower(envOr("ADMIN_USERNAME", "admin")) + email := strings.ToLower(envOr("ADMIN_EMAIL", "admin@example.com")) + password := os.Getenv("ADMIN_PASSWORD") + if password == "" { + if strings.EqualFold(envOr("DEV_MODE", "false"), "false") { + return fmt.Errorf("ADMIN_PASSWORD must be set when DEV_MODE is false") + } + password = "dev-only-admin-password" + } + + // Never re-apply ADMIN_PASSWORD (or profile fields) on restart — that undoes + // password changes made in the admin UI. Only create when missing. + if _, err := st.GetUserByEmail(ctx, email); err == nil { + return nil + } + if _, err := st.GetUserByUsername(ctx, username); err == nil { + return nil + } + + hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost) + if err != nil { + return err + } + if _, err := st.CreateUser(ctx, name, email, username, string(hash), "", "", "", models.RoleAdmin); err != nil { + return err + } + log.Printf("bootstrap admin ready: %s (%s)", email, username) + return nil +} + +func envOr(key, fallback string) string { + if v := strings.TrimSpace(os.Getenv(key)); v != "" { + return v + } + return fallback +} + +func seedCurriculum(ctx context.Context, st *store.Store) error { + chapters := []struct { + title, desc, icon string + order int + }{ + {"هفتهٔ ۱ · مبانی", "ترمینال، گیت و ذهنیت مهندسی.", "terminal", 1}, + {"هفتهٔ ۲ · سامانه‌های اصلی", "سازوکار درونی کد و ماشین‌ها.", "chip", 2}, + {"هفتهٔ ۳ · ساخت و انتشار", "کار روی پروژه، استقرار و بازبینی.", "rocket", 3}, + } + var chapterIDs []int64 + for _, c := range chapters { + ch := &models.Chapter{Title: c.title, Description: c.desc, Icon: c.icon, SortOrder: c.order} + id, err := st.CreateChapter(ctx, ch) + if err != nil { + return err + } + chapterIDs = append(chapterIDs, id) + } + + type lessonDef struct { + chapter int + title string + desc string + requires int64 // 0 = root + x int + y int + xp int + } + defs := []lessonDef{ + {0, "به پرگار خوش آمدید", "فلسفهٔ دوره، برنامهٔ زمانی و پیمان نظم.", 0, 0, 0, 50}, + {0, "مبانی شل و گیت", "کار با ترمینال و کنترل نسخه در نخستین مخزن شما.", 1, 1, 0, 60}, + {0, "مبانی شبکه", "رایانه‌ها چگونه با هم ارتباط می‌گیرند: TCP/IP، DNS و HTTP.", 1, 1, 1, 60}, + {1, "ساختمان داده ۱۰۱", "آرایه‌ها، هش‌ها و هنر انتخاب گزینهٔ درست.", 2, 2, 0, 70}, + {1, "راه‌اندازی ابزارها", "فایل‌های تنظیمات، ویرایشگرها و محیط توسعهٔ تکرارپذیر.", 2, 0, 2, 70}, + {2, "پروژه: انتشار یک CLI", "از مشخصات پروژه تا ابزار خط فرمان آمادهٔ انتشار.", 4, 3, 1, 100}, + } + + lessonIDs := map[int64]int64{} + for i, d := range defs { + var req *int64 + if d.requires != 0 { + rid := lessonIDs[d.requires] + req = &rid + } + l := &models.Lesson{ + ChapterID: chapterIDs[d.chapter], + Title: d.title, + Description: d.desc, + VideoKey: "sample.mp4", + DurationSeconds: 45, + CompletionThresholdPct: 85, + RequiresLessonID: req, + X: d.x, Y: d.y, + SortOrder: d.y, + XPReward: d.xp, + IsActive: true, + } + id, err := st.CreateLesson(ctx, l) + if err != nil { + return err + } + lessonIDs[int64(i)+1] = id + if err := seedQuestions(ctx, st, id, d.title); err != nil { + return err + } + } + return nil +} + +func seedQuestions(ctx context.Context, st *store.Store, lessonID int64, lessonTitle string) error { + templates := []struct { + q string + opts []string + answer int + explain string + }{ + {"درس " + quote(lessonTitle) + " چه چیزی به شما می‌آموزد؟", []string{ + "تماشای منفعلانهٔ محتوا", + "مهارت عملی و تأییدشده از مسیر نظم", + "پرش از بخش‌ها در هر زمان ممکن", + "هیچ‌چیز؛ این فقط یک جای خالی است", + }, 1, "پرگار بر توانمندی عملی و قابل‌سنجش تمرکز دارد، نه تماشای منفعلانه."}, + {"چرا پیش از باز شدن آزمون‌ها، زمان تماشای تأییدشده لازم است؟", []string{ + "تا عمداً سرعت پیشرفت هنرجو کم شود", + "تا مطمئن شویم واقعاً با محتوای درس درگیر شده‌اید", + "چون ویدیوها خسته‌کننده‌اند", + "برای فروش جان بیشتر", + }, 1, "زمان تماشای تأییدشده نشان می‌دهد پیش از آزمون واقعاً با محتوا کار کرده‌اید."}, + {"وقتی هر ۳ جان خود را از دست بدهید چه اتفاقی می‌افتد؟", []string{ + "هیچ؛ جان‌ها بلافاصله برمی‌گردند", + "امتیاز اضافه می‌گیرید", + "حساب شما تا بررسی متصدی محدود می‌شود", + "یک درس مخفی برایتان باز می‌شود", + }, 2, "با تمام شدن جان‌ها، حساب تا بررسی متصدی محدود می‌شود."}, + } + for pos, t := range templates { + q := &models.MCQQuestion{ + LessonID: lessonID, + Position: pos + 1, + Question: t.q, + Options: t.opts, + AnswerIndex: t.answer, + Explanation: t.explain, + } + if _, err := st.CreateQuestion(ctx, q); err != nil { + return err + } + } + return nil +} + +func quote(s string) string { + return "\"" + s + "\"" +} + +func seedEvents(ctx context.Context, st *store.Store) error { + now := time.Now() + events := []models.Event{ + { + Title: "آغاز زنده — هفتهٔ ۱", Description: "جلسهٔ معارفه، نقشهٔ مسیر دوره و انتظارات.", + EventType: "workshop", ExternalURL: "https://meet.google.com/pargar-kickoff", + StartsAt: now.Add(2 * time.Hour), EndsAt: now.Add(3 * time.Hour), IsActive: true, + }, + { + Title: "ساعت گفتگوی منتور", Description: "برای پرسش، بازبینی کد و رفع محدودیت حساب وارد شوید.", + EventType: "meet", ExternalURL: "https://meet.google.com/pargar-office", + StartsAt: now.Add(24 * time.Hour), EndsAt: now.Add(25 * time.Hour), IsActive: true, + }, + { + Title: "کارگاه پروژه: انتشار یک CLI", Description: "ساخت زندهٔ نخستین پروژه با همراهی منتور.", + EventType: "zoom", ExternalURL: "https://meet.google.com/pargar-cli", + StartsAt: now.Add(96 * time.Hour), EndsAt: now.Add(98 * time.Hour), IsActive: true, + }, + } + adminEmail := strings.ToLower(envOr("ADMIN_EMAIL", "admin@example.com")) + for i := range events { + e := &events[i] + admin, _ := st.GetUserByEmail(ctx, adminEmail) + if admin != nil { + id := admin.ID + e.CreatedBy = &id + } + if _, err := st.CreateEvent(ctx, e); err != nil { + return err + } + } + log.Println("seeded curriculum and events") + return nil +} diff --git a/backend/internal/service/hearts.go b/backend/internal/service/hearts.go new file mode 100644 index 0000000..63f6819 --- /dev/null +++ b/backend/internal/service/hearts.go @@ -0,0 +1,94 @@ +package service + +import ( + "context" + "time" + + "pargar/backend/internal/config" + "pargar/backend/internal/models" + "pargar/backend/internal/store" +) + +// HeartsService manages the fixed 3-heart pool and lockout. +// Hearts do not regenerate over time — only a mentor/admin unlock restores them. +type HeartsService struct { + store *store.Store +} + +func NewHeartsService(st *store.Store, _ *config.Config) *HeartsService { + return &HeartsService{store: st} +} + +// RefreshHearts is a no-op kept for call-site compatibility. +// The game has no timed heart refill; the stored count is authoritative. +func (h *HeartsService) RefreshHearts(_ context.Context, u *models.User) (*models.User, error) { + return u, nil +} + +// ConsumeHearts decrements hearts for wrong answers; locks the account at 0. +// Admin and mentor accounts have infinite hearts and are never locked by this path. +// Returns the updated user and how many hearts were lost. +func (h *HeartsService) ConsumeHearts(ctx context.Context, u *models.User, loss int) (*models.User, int, error) { + if u.Role.IsStaff() { + return u, 0, nil + } + if loss < 0 { + loss = 0 + } + prev := u.Hearts + cur := prev - loss + if cur < 0 { + cur = 0 + } + actualLost := prev - cur + if err := h.store.SetHearts(ctx, u.ID, cur); err != nil { + return nil, 0, err + } + u.Hearts = cur + if cur == 0 && !u.IsLocked { + locked, err := h.store.LockUser(ctx, u.ID) + if err != nil { + return nil, 0, err + } + return locked, actualLost, nil + } + return u, actualLost, nil +} + +// Streak logic keyed on activity date. +func UpdateStreak(ctx context.Context, st *store.Store, u *models.User) (*models.User, error) { + today := time.Now() + current, err := st.GetUserByID(ctx, u.ID) + if err != nil { + return nil, err + } + streak := current.StreakCurrent + longest := current.StreakLongest + if current.LastActivityDate != nil { + last := time.Date(current.LastActivityDate.Year(), current.LastActivityDate.Month(), current.LastActivityDate.Day(), 0, 0, 0, 0, time.UTC) + todayDate := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, time.UTC) + diff := int(todayDate.Sub(last).Hours() / 24) + switch { + case diff == 0: + // already active today + return current, nil + case diff == 1: + streak++ + default: + streak = 1 + } + } else { + streak = 1 + } + if streak > longest { + longest = streak + } + lastDate := today + if err := st.UpdateStreakOnly(ctx, current.ID, streak, longest, &lastDate); err != nil { + return nil, err + } + current.StreakCurrent = streak + current.StreakLongest = longest + current.LastActivityDate = &lastDate + return current, nil +} diff --git a/backend/internal/service/hearts_staff_test.go b/backend/internal/service/hearts_staff_test.go new file mode 100644 index 0000000..cfe4bd8 --- /dev/null +++ b/backend/internal/service/hearts_staff_test.go @@ -0,0 +1,28 @@ +package service + +import ( + "context" + "testing" + + "pargar/backend/internal/models" +) + +func TestConsumeHeartsSkipsStaff(t *testing.T) { + h := &HeartsService{} + for _, role := range []models.Role{models.RoleAdmin, models.RoleMentor} { + u := &models.User{ID: 1, Role: role, Hearts: 1, IsLocked: false} + out, lost, err := h.ConsumeHearts(context.Background(), u, 3) + if err != nil { + t.Fatalf("%s: unexpected err %v", role, err) + } + if lost != 0 { + t.Fatalf("%s: expected 0 hearts lost, got %d", role, lost) + } + if out.Hearts != 1 { + t.Fatalf("%s: hearts should stay 1, got %d", role, out.Hearts) + } + if out.IsLocked { + t.Fatalf("%s: staff must not lock", role) + } + } +} diff --git a/backend/internal/service/notify.go b/backend/internal/service/notify.go new file mode 100644 index 0000000..d4d6e0e --- /dev/null +++ b/backend/internal/service/notify.go @@ -0,0 +1,61 @@ +package service + +import ( + "context" + + "pargar/backend/internal/models" + "pargar/backend/internal/store" +) + +// Pusher delivers realtime messages to a user's connected websockets. +type Pusher interface { + Push(userID int64, msg any) +} + +// Notifier persists in-app notifications and pushes them live. +type Notifier struct { + store *store.Store + pusher Pusher +} + +func NewNotifier(st *store.Store, pusher Pusher) *Notifier { + return &Notifier{store: st, pusher: pusher} +} + +func (n *Notifier) Notify(ctx context.Context, userID int64, category, typ, title, body, route string, data any) error { + enabled, err := n.store.PrefEnabled(ctx, userID, category) + if err != nil || !enabled { + return nil + } + notif := &models.Notification{ + UserID: userID, + Category: category, + Type: typ, + Title: title, + Body: body, + Route: route, + } + if data != nil { + notif.Data = mustJSON(data) + } + if err := n.store.CreateNotification(ctx, notif); err != nil { + return err + } + if n.pusher != nil { + n.pusher.Push(userID, map[string]any{ + "type": "notification", + "unread": true, + "item": notif, + }) + } + return nil +} + +func (n *Notifier) NotifyMany(ctx context.Context, userIDs []int64, category, typ, title, body, route string, data any) error { + for _, uid := range userIDs { + if err := n.Notify(ctx, uid, category, typ, title, body, route, data); err != nil { + return err + } + } + return nil +} diff --git a/backend/internal/service/util.go b/backend/internal/service/util.go new file mode 100644 index 0000000..1025d04 --- /dev/null +++ b/backend/internal/service/util.go @@ -0,0 +1,21 @@ +package service + +import ( + "encoding/json" + "fmt" + "time" +) + +func mustJSON(v any) json.RawMessage { + b, err := json.Marshal(v) + if err != nil { + return json.RawMessage(`{}`) + } + return b +} + +// ISOWeek returns the ISO-8601 week key like "2026-W37". +func ISOWeek(t time.Time) string { + year, week := t.ISOWeek() + return fmt.Sprintf("%d-W%02d", year, week) +} diff --git a/backend/internal/service/watch.go b/backend/internal/service/watch.go new file mode 100644 index 0000000..742260b --- /dev/null +++ b/backend/internal/service/watch.go @@ -0,0 +1,112 @@ +package service + +import ( + "context" + "errors" + "time" + + "pargar/backend/internal/models" + "pargar/backend/internal/store" +) + +var ( + ErrLocked = errors.New("حساب محدود شده است") + ErrLessonLocked = errors.New("درس قفل است") + ErrQuizNotUnlocked = errors.New("آزمون هنوز باز نشده است") + ErrAlreadyPassed = errors.New("درس قبلاً تکمیل شده است") +) + +// Max seconds of watch progress accepted per heartbeat (client interval is 10s). +const maxHeartbeatDelta = 15.0 + +// wallClockSlack is added to elapsed wall time when capping delta (timer jitter). +const wallClockSlack = 1.5 + +// WatchService validates heartbeats and computes verified watch progress. +type WatchService struct { + store *store.Store +} + +func NewWatchService(st *store.Store) *WatchService { + return &WatchService{store: st} +} + +// Heartbeat records verified playback progress. +// position = playback position in seconds; delta = seconds actually played since last ping. +func (w *WatchService) Heartbeat(ctx context.Context, userID int64, lesson *models.Lesson, position, delta float64, seq int) (*models.LessonProgress, error) { + if position < 0 { + position = 0 + } + dur := float64(lesson.DurationSeconds) + if dur <= 0 { + dur = 1 + } + // Clamp position within duration + if position > dur { + position = dur + } + if delta < 0 { + delta = 0 + } + if delta > maxHeartbeatDelta { + delta = maxHeartbeatDelta + } + if delta > dur { + delta = dur + } + + session, err := w.store.GetOrCreateSession(ctx, userID, lesson.ID) + if err != nil { + return nil, err + } + // Cap by wall clock so speedup / forged deltas cannot outrun real time. + elapsed := time.Since(session.LastHeartbeatAt).Seconds() + if elapsed < 0 { + elapsed = 0 + } + wallCap := elapsed + wallClockSlack + if delta > wallCap { + delta = wallCap + } + inserted, err := w.store.InsertHeartbeat(ctx, session, position, delta, seq) + if err != nil { + return nil, err + } + if err := w.store.UpdateSessionPos(ctx, session.ID, position); err != nil { + return nil, err + } + + pg, err := w.store.GetProgress(ctx, userID, lesson.ID) + if err != nil { + pg = &models.LessonProgress{UserID: userID, LessonID: lesson.ID} + } + // Only credit progress when the heartbeat row was newly inserted (prevents seq replay). + if inserted { + pg.WatchedSeconds += delta + } + if pg.WatchedSeconds > dur { + pg.WatchedSeconds = dur + } + pg.WatchedPct = pg.WatchedSeconds / dur * 100 + pg.LastPosition = position + + // mark quiz unlocked once threshold met + if !pg.QuizUnlocked && pg.WatchedPct >= float64(lesson.CompletionThresholdPct) { + pg.QuizUnlocked = true + if err := w.store.SetQuizUnlocked(ctx, userID, lesson.ID); err != nil { + return nil, err + } + } + if err := w.store.UpsertProgress(ctx, pg); err != nil { + return nil, err + } + return pg, nil +} + +func (w *WatchService) ResumePosition(ctx context.Context, userID, lessonID int64) (float64, error) { + pg, err := w.store.GetProgress(ctx, userID, lessonID) + if err != nil { + return 0, nil + } + return pg.LastPosition, nil +} diff --git a/backend/internal/service/watch_test.go b/backend/internal/service/watch_test.go new file mode 100644 index 0000000..bc46183 --- /dev/null +++ b/backend/internal/service/watch_test.go @@ -0,0 +1,50 @@ +package service + +import ( + "testing" + "time" + + "pargar/backend/internal/models" +) + +func TestHeartbeatCapsDelta(t *testing.T) { + lesson := &models.Lesson{ID: 1, DurationSeconds: 100, CompletionThresholdPct: 85} + delta := 999.0 + if delta > maxHeartbeatDelta { + delta = maxHeartbeatDelta + } + if delta > float64(lesson.DurationSeconds) { + delta = float64(lesson.DurationSeconds) + } + if delta != maxHeartbeatDelta { + t.Fatalf("expected cap %v, got %v", maxHeartbeatDelta, delta) + } +} + +func TestHeartbeatCapsDeltaByWallClock(t *testing.T) { + last := time.Now().Add(-2 * time.Second) + elapsed := time.Since(last).Seconds() + wallCap := elapsed + wallClockSlack + delta := 15.0 + if delta > wallCap { + delta = wallCap + } + if delta > maxHeartbeatDelta { + delta = maxHeartbeatDelta + } + if delta > 5 { + t.Fatalf("wall-clock cap should keep delta near ~3.5s, got %v", delta) + } + if delta < 2 { + t.Fatalf("wall-clock cap too aggressive: %v", delta) + } +} + +func TestMaxHeartbeatDeltaConstant(t *testing.T) { + if maxHeartbeatDelta != 15.0 { + t.Fatalf("maxHeartbeatDelta changed unexpectedly: %v", maxHeartbeatDelta) + } + if wallClockSlack < 1 { + t.Fatalf("wallClockSlack too small: %v", wallClockSlack) + } +} diff --git a/backend/internal/store/blacklist.go b/backend/internal/store/blacklist.go new file mode 100644 index 0000000..4600594 --- /dev/null +++ b/backend/internal/store/blacklist.go @@ -0,0 +1,281 @@ +package store + +import ( + "context" + "errors" + "strings" + + "github.com/jackc/pgx/v5" + + "pargar/backend/internal/models" +) + +var ( + ErrPhoneBlacklisted = errors.New("phone is blacklisted") + ErrBlacklistNotFound = errors.New("blacklist entry not found") +) + +const blacklistCols = `id, phone, created_by, created_at` + +func scanBlacklist(row rowScanner) (*models.RegistrationPhoneBlacklist, error) { + var e models.RegistrationPhoneBlacklist + err := row.Scan(&e.ID, &e.Phone, &e.CreatedBy, &e.CreatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrBlacklistNotFound + } + if err != nil { + return nil, err + } + return &e, nil +} + +// IsPhoneBlacklisted reports whether phone is on the registration blacklist. +func (s *Store) IsPhoneBlacklisted(ctx context.Context, phone string) (bool, error) { + var exists bool + err := s.pool.QueryRow(ctx, ` + SELECT EXISTS(SELECT 1 FROM registration_phone_blacklist WHERE phone=$1)`, phone).Scan(&exists) + return exists, err +} + +func isPhoneBlacklistedTx(ctx context.Context, tx pgx.Tx, phone string) (bool, error) { + var exists bool + err := tx.QueryRow(ctx, ` + SELECT EXISTS(SELECT 1 FROM registration_phone_blacklist WHERE phone=$1)`, phone).Scan(&exists) + return exists, err +} + +// BlacklistImportResult is the outcome of a blacklist import batch. +type BlacklistImportResult struct { + Inserted int + Skipped int + RemovedWhitelist int + Rejected int + RejectedPhones []string + WhitelistedConflicts []string +} + +// ImportBlacklistPhones inserts phones onto the blacklist. +// Already-blacklisted phones are skipped. +// Available whitelist entries are reported in WhitelistedConflicts unless resolveConflicts +// is true, in which case they are removed from the whitelist then blacklisted. +// Phones already consumed as users (whitelist consumed or existing user) are rejected. +func (s *Store) ImportBlacklistPhones(ctx context.Context, phones []string, createdBy int64, resolveConflicts bool) (BlacklistImportResult, error) { + var result BlacklistImportResult + if len(phones) == 0 { + return result, nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return result, err + } + defer tx.Rollback(ctx) + + for _, phone := range phones { + phone = strings.TrimSpace(phone) + if phone == "" { + continue + } + + var wlStatus string + wlErr := tx.QueryRow(ctx, ` + SELECT status FROM registration_phone_whitelist WHERE phone=$1`, phone).Scan(&wlStatus) + if wlErr != nil && !errors.Is(wlErr, pgx.ErrNoRows) { + return BlacklistImportResult{}, wlErr + } + onWhitelist := wlErr == nil + + if onWhitelist && wlStatus == string(models.WhitelistConsumed) { + result.Rejected++ + result.RejectedPhones = append(result.RejectedPhones, phone) + continue + } + + var userExists bool + if err := tx.QueryRow(ctx, ` + SELECT EXISTS(SELECT 1 FROM users WHERE phone=$1)`, phone).Scan(&userExists); err != nil { + return BlacklistImportResult{}, err + } + if userExists { + result.Rejected++ + result.RejectedPhones = append(result.RejectedPhones, phone) + continue + } + + if onWhitelist && wlStatus == string(models.WhitelistAvailable) { + if !resolveConflicts { + result.WhitelistedConflicts = append(result.WhitelistedConflicts, phone) + continue + } + tag, err := tx.Exec(ctx, ` + DELETE FROM registration_phone_whitelist + WHERE phone = $1 AND status = 'available'`, phone) + if err != nil { + return BlacklistImportResult{}, err + } + if tag.RowsAffected() > 0 { + result.RemovedWhitelist++ + } + } + + tag, err := tx.Exec(ctx, ` + INSERT INTO registration_phone_blacklist (phone, created_by) + VALUES ($1, $2) + ON CONFLICT (phone) DO NOTHING`, phone, createdBy) + if err != nil { + return BlacklistImportResult{}, err + } + if tag.RowsAffected() == 0 { + result.Skipped++ + } else { + result.Inserted++ + } + } + if err := assertPhoneListsExclusiveTx(ctx, tx); err != nil { + return BlacklistImportResult{}, err + } + if err := tx.Commit(ctx); err != nil { + return BlacklistImportResult{}, err + } + return result, nil +} + +// assertPhoneListsExclusiveTx fails if any phone appears on both whitelist and blacklist. +func assertPhoneListsExclusiveTx(ctx context.Context, tx pgx.Tx) error { + var n int + if err := tx.QueryRow(ctx, ` + SELECT COUNT(*) FROM registration_phone_whitelist w + INNER JOIN registration_phone_blacklist b ON b.phone = w.phone`).Scan(&n); err != nil { + return err + } + if n > 0 { + return errors.New("phone cannot be on whitelist and blacklist at the same time") + } + return nil +} + +func (s *Store) ListBlacklistPhones(ctx context.Context, query string, limit, offset int) ([]models.RegistrationPhoneBlacklist, int64, error) { + if limit <= 0 { + limit = 10 + } + if limit > 100 { + limit = 100 + } + if offset < 0 { + offset = 0 + } + query = strings.TrimSpace(query) + var total int64 + if err := s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM registration_phone_blacklist + WHERE ($1 = '' OR phone ILIKE '%'||$1||'%')`, query).Scan(&total); err != nil { + return nil, 0, err + } + rows, err := s.pool.Query(ctx, ` + SELECT `+blacklistCols+` + FROM registration_phone_blacklist + WHERE ($1 = '' OR phone ILIKE '%'||$1||'%') + ORDER BY created_at DESC + LIMIT $2 OFFSET $3`, query, limit, offset) + if err != nil { + return nil, 0, err + } + defer rows.Close() + var out []models.RegistrationPhoneBlacklist + for rows.Next() { + e, err := scanBlacklist(rows) + if err != nil { + return nil, 0, err + } + out = append(out, *e) + } + return out, total, rows.Err() +} + +// DeleteBlacklistPhone removes a blacklist entry by id. +func (s *Store) DeleteBlacklistPhone(ctx context.Context, id int64) error { + tag, err := s.pool.Exec(ctx, ` + DELETE FROM registration_phone_blacklist WHERE id = $1`, id) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrBlacklistNotFound + } + return nil +} + +const blacklistAttemptCols = `id, phone, path, invite_id, ip, user_agent, + attempted_username, attempted_email, attempted_at` + +func scanBlacklistAttempt(row rowScanner) (*models.RegistrationBlacklistAttempt, error) { + var e models.RegistrationBlacklistAttempt + err := row.Scan( + &e.ID, &e.Phone, &e.Path, &e.InviteID, &e.IP, &e.UserAgent, + &e.AttemptedUsername, &e.AttemptedEmail, &e.AttemptedAt, + ) + if err != nil { + return nil, err + } + return &e, nil +} + +// RecordBlacklistAttempt inserts an audit row when a blacklisted phone tries to register. +func (s *Store) RecordBlacklistAttempt( + ctx context.Context, + phone, path string, + inviteID *int64, + ip, userAgent, username, email string, +) (*models.RegistrationBlacklistAttempt, error) { + if path != "public" && path != "invite" { + path = "public" + } + row := s.pool.QueryRow(ctx, ` + INSERT INTO registration_blacklist_attempts + (phone, path, invite_id, ip, user_agent, attempted_username, attempted_email) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING `+blacklistAttemptCols, + phone, path, inviteID, ip, userAgent, username, email, + ) + return scanBlacklistAttempt(row) +} + +// ListBlacklistAttempts returns recent attempts, optionally filtered by q (phone/username/email). +func (s *Store) ListBlacklistAttempts(ctx context.Context, query string, limit, offset int) ([]models.RegistrationBlacklistAttempt, int64, error) { + if limit <= 0 { + limit = 10 + } + if limit > 100 { + limit = 100 + } + if offset < 0 { + offset = 0 + } + query = strings.TrimSpace(query) + where := `($1 = '' OR phone ILIKE '%'||$1||'%' + OR COALESCE(attempted_username, '') ILIKE '%'||$1||'%' + OR COALESCE(attempted_email, '') ILIKE '%'||$1||'%')` + var total int64 + if err := s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM registration_blacklist_attempts WHERE `+where, query).Scan(&total); err != nil { + return nil, 0, err + } + rows, err := s.pool.Query(ctx, ` + SELECT `+blacklistAttemptCols+` + FROM registration_blacklist_attempts + WHERE `+where+` + ORDER BY attempted_at DESC + LIMIT $2 OFFSET $3`, query, limit, offset) + if err != nil { + return nil, 0, err + } + defer rows.Close() + var out []models.RegistrationBlacklistAttempt + for rows.Next() { + e, err := scanBlacklistAttempt(rows) + if err != nil { + return nil, 0, err + } + out = append(out, *e) + } + return out, total, rows.Err() +} diff --git a/backend/internal/store/challenges.go b/backend/internal/store/challenges.go new file mode 100644 index 0000000..501227e --- /dev/null +++ b/backend/internal/store/challenges.go @@ -0,0 +1,135 @@ +package store + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + + "pargar/backend/internal/models" +) + +func (s *Store) CreateChallenge(ctx context.Context, c *models.Challenge) (int64, error) { + err := s.pool.QueryRow(ctx, ` + INSERT INTO challenges (title, description, target_xp, starts_at, ends_at, is_active, created_by) + VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id, created_at`, + c.Title, c.Description, c.TargetXP, c.StartsAt, c.EndsAt, c.IsActive, c.CreatedBy, + ).Scan(&c.ID, &c.CreatedAt) + return c.ID, err +} + +func (s *Store) UpdateChallenge(ctx context.Context, id int64, c *models.Challenge) error { + tag, err := s.pool.Exec(ctx, ` + UPDATE challenges SET title=$1, description=$2, target_xp=$3, starts_at=$4, ends_at=$5, is_active=$6 + WHERE id=$7`, + c.Title, c.Description, c.TargetXP, c.StartsAt, c.EndsAt, c.IsActive, id) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +func (s *Store) DeleteChallenge(ctx context.Context, id int64) error { + tag, err := s.pool.Exec(ctx, `DELETE FROM challenges WHERE id=$1`, id) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +func (s *Store) ListChallengesAdmin(ctx context.Context) ([]models.Challenge, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, title, description, target_xp, starts_at, ends_at, is_active, created_at + FROM challenges ORDER BY starts_at DESC LIMIT 200`) + if err != nil { + return nil, err + } + defer rows.Close() + return scanChallenges(rows) +} + +// ListActiveChallengesForUser returns challenges currently in window (active + now in range) with progress. +func (s *Store) ListActiveChallengesForUser(ctx context.Context, userID int64) ([]models.Challenge, error) { + rows, err := s.pool.Query(ctx, ` + SELECT c.id, c.title, c.description, c.target_xp, c.starts_at, c.ends_at, c.is_active, c.created_at, + COALESCE(p.xp, 0) AS my_xp + FROM challenges c + LEFT JOIN challenge_progress p ON p.challenge_id = c.id AND p.user_id = $1 + WHERE c.is_active AND c.starts_at <= now() AND c.ends_at >= now() + ORDER BY c.ends_at ASC`, userID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.Challenge + for rows.Next() { + var c models.Challenge + if err := rows.Scan(&c.ID, &c.Title, &c.Description, &c.TargetXP, &c.StartsAt, &c.EndsAt, &c.IsActive, &c.CreatedAt, &c.MyXP); err != nil { + return nil, err + } + c.Completed = c.MyXP >= c.TargetXP + out = append(out, c) + } + if out == nil { + out = []models.Challenge{} + } + return out, rows.Err() +} + +// AddChallengeProgress adds XP toward all active challenges in the current window. +func (s *Store) AddChallengeProgress(ctx context.Context, userID int64, amount int) error { + if amount <= 0 { + return nil + } + _, err := s.pool.Exec(ctx, ` + INSERT INTO challenge_progress (challenge_id, user_id, xp, updated_at) + SELECT c.id, $1, $2, now() + FROM challenges c + WHERE c.is_active AND c.starts_at <= now() AND c.ends_at >= now() + ON CONFLICT (challenge_id, user_id) DO UPDATE + SET xp = challenge_progress.xp + EXCLUDED.xp, + updated_at = now()`, userID, amount) + return err +} + +func scanChallenges(rows pgx.Rows) ([]models.Challenge, error) { + var out []models.Challenge + for rows.Next() { + var c models.Challenge + if err := rows.Scan(&c.ID, &c.Title, &c.Description, &c.TargetXP, &c.StartsAt, &c.EndsAt, &c.IsActive, &c.CreatedAt); err != nil { + return nil, err + } + out = append(out, c) + } + if out == nil { + out = []models.Challenge{} + } + return out, rows.Err() +} + +func (s *Store) GetChallenge(ctx context.Context, id int64) (*models.Challenge, error) { + row := s.pool.QueryRow(ctx, ` + SELECT id, title, description, target_xp, starts_at, ends_at, is_active, created_at + FROM challenges WHERE id=$1`, id) + var c models.Challenge + err := row.Scan(&c.ID, &c.Title, &c.Description, &c.TargetXP, &c.StartsAt, &c.EndsAt, &c.IsActive, &c.CreatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return &c, nil +} + +// ChallengeWindowOK is a small helper for tests / validation. +func ChallengeWindowOK(starts, ends time.Time) bool { + return !starts.IsZero() && !ends.IsZero() && !ends.Before(starts) +} diff --git a/backend/internal/store/chat_export.go b/backend/internal/store/chat_export.go new file mode 100644 index 0000000..b9de428 --- /dev/null +++ b/backend/internal/store/chat_export.go @@ -0,0 +1,243 @@ +package store + +import ( + "context" + "errors" + "strings" + "time" + "unicode/utf8" + + "github.com/jackc/pgx/v5" + + "pargar/backend/internal/models" +) + +var ( + ErrChatExportTokenNotFound = errors.New("chat export token not found") + ErrChatExportTokenRevoked = errors.New("chat export token revoked") + ErrChatExportTokenExpired = errors.New("chat export token expired") + ErrChatExportTokenUsed = errors.New("chat export token already used") + ErrChatExportTokenInvalid = errors.New("chat export token invalid") +) + +const chatExportTokenCols = `id, token_hash, token_enc, label, expires_at, revoked_at, used_at, + created_by, created_at, last_used_at` + +// ChatExportPair is one student↔mentor conversation found in chat_messages. +type ChatExportPair struct { + StudentID int64 + StudentName string + MentorID int64 + MentorName string +} + +// ChatExportStats is an approximate size summary for the full chat archive. +type ChatExportStats struct { + ConversationCount int64 `json:"conversationCount"` + MessageCount int64 `json:"messageCount"` + AttachmentBytes int64 `json:"attachmentBytes"` + BodyBytes int64 `json:"bodyBytes"` + // EstimatedBytes is a rough upper-bound for the ZIP (attachments + compressed text overhead). + EstimatedBytes int64 `json:"estimatedBytes"` +} + +// ChatExportStats returns approximate totals for the project-wide chat archive. +func (s *Store) ChatExportStats(ctx context.Context) (*ChatExportStats, error) { + var st ChatExportStats + err := s.pool.QueryRow(ctx, ` + SELECT + COALESCE((SELECT COUNT(*) FROM (SELECT DISTINCT user_id, mentor_id FROM chat_messages) p), 0), + COUNT(*), + COALESCE(SUM(COALESCE(attachment_size, 0)), 0), + COALESCE(SUM(OCTET_LENGTH(COALESCE(body, ''))), 0) + FROM chat_messages`).Scan( + &st.ConversationCount, + &st.MessageCount, + &st.AttachmentBytes, + &st.BodyBytes, + ) + if err != nil { + return nil, err + } + // Text/JSON compresses well in ZIP; media usually does not. Bias toward a safe upper bound. + const perMessageJSON int64 = 160 + const perConversationJSON int64 = 120 + const zipEnvelope int64 = 4096 + textPacked := st.BodyBytes/2 + st.MessageCount*perMessageJSON + st.ConversationCount*perConversationJSON + zipEnvelope + st.EstimatedBytes = st.AttachmentBytes + textPacked + return &st, nil +} + +// ListChatExportPairs returns every distinct conversation pair with display names. +func (s *Store) ListChatExportPairs(ctx context.Context) ([]ChatExportPair, error) { + rows, err := s.pool.Query(ctx, ` + SELECT m.user_id, COALESCE(su.name, ''), m.mentor_id, COALESCE(mu.name, '') + FROM ( + SELECT DISTINCT user_id, mentor_id FROM chat_messages + ) m + LEFT JOIN users su ON su.id = m.user_id + LEFT JOIN users mu ON mu.id = m.mentor_id + ORDER BY m.user_id, m.mentor_id`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []ChatExportPair + for rows.Next() { + var p ChatExportPair + if err := rows.Scan(&p.StudentID, &p.StudentName, &p.MentorID, &p.MentorName); err != nil { + return nil, err + } + out = append(out, p) + } + return out, rows.Err() +} + +func chatExportTokenActive(tok *models.ChatExportToken, now time.Time) bool { + return tok != nil && tok.RevokedAt == nil && tok.UsedAt == nil && tok.ExpiresAt.After(now) +} + +func scanChatExportToken(row rowScanner) (*models.ChatExportToken, error) { + var tok models.ChatExportToken + var tokenHash, tokenEnc string + err := row.Scan( + &tok.ID, &tokenHash, &tokenEnc, &tok.Label, &tok.ExpiresAt, &tok.RevokedAt, &tok.UsedAt, + &tok.CreatedBy, &tok.CreatedAt, &tok.LastUsedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrChatExportTokenNotFound + } + if err != nil { + return nil, err + } + tok.Active = chatExportTokenActive(&tok, time.Now().UTC()) + // Stash enc on Token for decrypt by callers; clear/replace before JSON if unused. + tok.Token = tokenEnc + return &tok, nil +} + +func (s *Store) CreateChatExportToken( + ctx context.Context, + tokenHash, tokenEnc, label string, + expiresAt time.Time, + createdBy int64, +) (*models.ChatExportToken, error) { + label = strings.TrimSpace(label) + if utf8.RuneCountInString(label) > 120 { + label = string([]rune(label)[:120]) + } + if expiresAt.IsZero() || !expiresAt.After(time.Now().UTC()) { + return nil, errors.New("expiresAt must be in the future") + } + row := s.pool.QueryRow(ctx, ` + INSERT INTO chat_export_tokens (token_hash, token_enc, label, expires_at, created_by) + VALUES ($1, $2, $3, $4, $5) + RETURNING `+chatExportTokenCols, + tokenHash, tokenEnc, label, expiresAt.UTC(), createdBy) + return scanChatExportToken(row) +} + +func (s *Store) ListChatExportTokens(ctx context.Context, limit int) ([]models.ChatExportToken, error) { + if limit <= 0 { + limit = 50 + } + if limit > 200 { + limit = 200 + } + rows, err := s.pool.Query(ctx, ` + SELECT `+chatExportTokenCols+` + FROM chat_export_tokens + ORDER BY created_at DESC + LIMIT $1`, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.ChatExportToken + for rows.Next() { + tok, err := scanChatExportToken(rows) + if err != nil { + return nil, err + } + out = append(out, *tok) + } + return out, rows.Err() +} + +func (s *Store) GetChatExportTokenByHash(ctx context.Context, tokenHash string) (*models.ChatExportToken, error) { + row := s.pool.QueryRow(ctx, ` + SELECT `+chatExportTokenCols+` + FROM chat_export_tokens WHERE token_hash=$1`, tokenHash) + return scanChatExportToken(row) +} + +func (s *Store) GetChatExportTokenByID(ctx context.Context, id int64) (*models.ChatExportToken, error) { + row := s.pool.QueryRow(ctx, ` + SELECT `+chatExportTokenCols+` + FROM chat_export_tokens WHERE id=$1`, id) + return scanChatExportToken(row) +} + +// ValidateChatExportToken returns nil when the token is active (not revoked, unused, not expired). +func ValidateChatExportToken(tok *models.ChatExportToken, now time.Time) error { + if tok == nil { + return ErrChatExportTokenNotFound + } + if tok.RevokedAt != nil { + return ErrChatExportTokenRevoked + } + if tok.UsedAt != nil { + return ErrChatExportTokenUsed + } + if !tok.ExpiresAt.After(now) { + return ErrChatExportTokenExpired + } + return nil +} + +// ConsumeChatExportToken marks a token used atomically (single-use download links). +func (s *Store) ConsumeChatExportToken(ctx context.Context, id int64) (*models.ChatExportToken, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE chat_export_tokens + SET used_at = now(), last_used_at = now() + WHERE id = $1 + AND revoked_at IS NULL + AND used_at IS NULL + AND expires_at > now() + RETURNING `+chatExportTokenCols, id) + tok, err := scanChatExportToken(row) + if err != nil { + if errors.Is(err, ErrChatExportTokenNotFound) { + existing, getErr := s.GetChatExportTokenByID(ctx, id) + if getErr != nil { + return nil, getErr + } + if err := ValidateChatExportToken(existing, time.Now().UTC()); err != nil { + return existing, err + } + return existing, ErrChatExportTokenInvalid + } + return nil, err + } + return tok, nil +} + +func (s *Store) RevokeChatExportToken(ctx context.Context, id int64) (*models.ChatExportToken, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE chat_export_tokens + SET revoked_at = now() + WHERE id = $1 AND revoked_at IS NULL + RETURNING `+chatExportTokenCols, id) + tok, err := scanChatExportToken(row) + if err != nil { + if errors.Is(err, ErrChatExportTokenNotFound) { + existing, getErr := s.GetChatExportTokenByID(ctx, id) + if getErr != nil { + return nil, getErr + } + return existing, ErrChatExportTokenInvalid + } + return nil, err + } + return tok, nil +} diff --git a/backend/internal/store/invites.go b/backend/internal/store/invites.go new file mode 100644 index 0000000..23504ff --- /dev/null +++ b/backend/internal/store/invites.go @@ -0,0 +1,450 @@ +package store + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "errors" + "io" + "strings" + "time" + "unicode/utf8" + + "github.com/jackc/pgx/v5" + + "pargar/backend/internal/models" +) + +var ( + ErrInviteNotFound = errors.New("invite not found") + ErrInviteInvalid = errors.New("invite invalid") + ErrInvitePaused = errors.New("invite paused") + ErrInviteRevoked = errors.New("invite revoked") + ErrInviteExhausted = errors.New("invite exhausted") + ErrInviteExpired = errors.New("invite expired") + ErrInviteMaxUsesTooLow = errors.New("maxUses below used_count") + ErrInviteMaxUsesRange = errors.New("maxUses out of range") +) + +const inviteCols = `id, token_hash, token_enc, label, max_uses, used_count, status, + expires_at, created_by, created_at` + +func HashInviteToken(token string) string { + sum := sha256.Sum256([]byte(token)) + return hex.EncodeToString(sum[:]) +} + +// EncryptInviteToken encrypts the raw invite token with AES-GCM keyed by secret. +func EncryptInviteToken(secret, plaintext string) (string, error) { + key := sha256.Sum256([]byte(secret)) + block, err := aes.NewCipher(key[:]) + if err != nil { + return "", err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return "", err + } + nonce := make([]byte, gcm.NonceSize()) + if _, err := io.ReadFull(rand.Reader, nonce); err != nil { + return "", err + } + out := gcm.Seal(nonce, nonce, []byte(plaintext), nil) + return base64.RawURLEncoding.EncodeToString(out), nil +} + +// DecryptInviteToken decrypts a token produced by EncryptInviteToken. +func DecryptInviteToken(secret, ciphertext string) (string, error) { + raw, err := base64.RawURLEncoding.DecodeString(ciphertext) + if err != nil { + return "", err + } + key := sha256.Sum256([]byte(secret)) + block, err := aes.NewCipher(key[:]) + if err != nil { + return "", err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return "", err + } + if len(raw) < gcm.NonceSize() { + return "", errors.New("ciphertext too short") + } + nonce, sealed := raw[:gcm.NonceSize()], raw[gcm.NonceSize():] + plain, err := gcm.Open(nil, nonce, sealed, nil) + if err != nil { + return "", err + } + return string(plain), nil +} + +// NewInviteToken returns 32 random bytes encoded as base64url (no padding). +func NewInviteToken() (string, error) { + b := make([]byte, 32) + if _, err := io.ReadFull(rand.Reader, b); err != nil { + return "", err + } + return base64.RawURLEncoding.EncodeToString(b), nil +} + +func inviteRemaining(maxUses, usedCount int, status models.InviteStatus) int { + // Admin still sees remaining capacity while paused; revoked/exhausted show 0. + if status != models.InviteActive && status != models.InvitePaused { + return 0 + } + rem := maxUses - usedCount + if rem < 0 { + return 0 + } + return rem +} + +func scanInvite(row rowScanner) (*models.RegistrationInvite, error) { + var inv models.RegistrationInvite + var tokenHash, tokenEnc, status string + err := row.Scan( + &inv.ID, &tokenHash, &tokenEnc, &inv.Label, &inv.MaxUses, &inv.UsedCount, &status, + &inv.ExpiresAt, &inv.CreatedBy, &inv.CreatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrInviteNotFound + } + if err != nil { + return nil, err + } + inv.Status = models.InviteStatus(status) + inv.Remaining = inviteRemaining(inv.MaxUses, inv.UsedCount, inv.Status) + // Stash enc on Token temporarily for decrypt by callers that need it; clear before JSON if unused. + inv.Token = tokenEnc + return &inv, nil +} + +func (s *Store) CreateRegistrationInvite( + ctx context.Context, + tokenHash, tokenEnc, label string, + maxUses int, + expiresAt *time.Time, + createdBy int64, +) (*models.RegistrationInvite, error) { + label = strings.TrimSpace(label) + if utf8.RuneCountInString(label) > 120 { + label = string([]rune(label)[:120]) + } + if maxUses < 1 { + return nil, errors.New("maxUses must be >= 1") + } + row := s.pool.QueryRow(ctx, ` + INSERT INTO registration_invites (token_hash, token_enc, label, max_uses, expires_at, created_by) + VALUES ($1, $2, $3, $4, $5, $6) + RETURNING `+inviteCols, + tokenHash, tokenEnc, label, maxUses, expiresAt, createdBy) + return scanInvite(row) +} + +func (s *Store) ListRegistrationInvites(ctx context.Context, query string, limit, offset int) ([]models.RegistrationInvite, int64, error) { + if limit <= 0 { + limit = 10 + } + if limit > 100 { + limit = 100 + } + if offset < 0 { + offset = 0 + } + query = strings.TrimSpace(query) + var total int64 + if err := s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM registration_invites + WHERE ($1 = '' OR COALESCE(label, '') ILIKE '%'||$1||'%')`, query).Scan(&total); err != nil { + return nil, 0, err + } + rows, err := s.pool.Query(ctx, ` + SELECT `+inviteCols+` + FROM registration_invites + WHERE ($1 = '' OR COALESCE(label, '') ILIKE '%'||$1||'%') + ORDER BY created_at DESC + LIMIT $2 OFFSET $3`, query, limit, offset) + if err != nil { + return nil, 0, err + } + defer rows.Close() + var out []models.RegistrationInvite + for rows.Next() { + inv, err := scanInvite(rows) + if err != nil { + return nil, 0, err + } + out = append(out, *inv) + } + return out, total, rows.Err() +} + +func (s *Store) GetRegistrationInviteByID(ctx context.Context, id int64) (*models.RegistrationInvite, error) { + row := s.pool.QueryRow(ctx, `SELECT `+inviteCols+` FROM registration_invites WHERE id=$1`, id) + return scanInvite(row) +} + +func (s *Store) GetRegistrationInviteByTokenHash(ctx context.Context, tokenHash string) (*models.RegistrationInvite, error) { + row := s.pool.QueryRow(ctx, `SELECT `+inviteCols+` FROM registration_invites WHERE token_hash=$1`, tokenHash) + return scanInvite(row) +} + +func (s *Store) RevokeRegistrationInvite(ctx context.Context, id int64) (*models.RegistrationInvite, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE registration_invites + SET status = 'revoked' + WHERE id = $1 AND status IN ('active', 'paused') + RETURNING `+inviteCols, id) + inv, err := scanInvite(row) + if err != nil { + if errors.Is(err, ErrInviteNotFound) { + existing, getErr := s.GetRegistrationInviteByID(ctx, id) + if getErr != nil { + return nil, getErr + } + return existing, ErrInviteInvalid + } + return nil, err + } + return inv, nil +} + +func (s *Store) PauseRegistrationInvite(ctx context.Context, id int64) (*models.RegistrationInvite, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE registration_invites + SET status = 'paused' + WHERE id = $1 AND status = 'active' + RETURNING `+inviteCols, id) + inv, err := scanInvite(row) + if err != nil { + if errors.Is(err, ErrInviteNotFound) { + existing, getErr := s.GetRegistrationInviteByID(ctx, id) + if getErr != nil { + return nil, getErr + } + return existing, ErrInviteInvalid + } + return nil, err + } + return inv, nil +} + +func (s *Store) ResumeRegistrationInvite(ctx context.Context, id int64) (*models.RegistrationInvite, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE registration_invites + SET status = 'active' + WHERE id = $1 AND status = 'paused' + AND used_count < max_uses + AND (expires_at IS NULL OR expires_at > now()) + RETURNING `+inviteCols, id) + inv, err := scanInvite(row) + if err != nil { + if errors.Is(err, ErrInviteNotFound) { + existing, getErr := s.GetRegistrationInviteByID(ctx, id) + if getErr != nil { + return nil, getErr + } + if existing.Status != models.InvitePaused { + return existing, ErrInviteInvalid + } + if existing.ExpiresAt != nil && !existing.ExpiresAt.After(time.Now()) { + return existing, ErrInviteExpired + } + if existing.UsedCount >= existing.MaxUses { + return existing, ErrInviteExhausted + } + return existing, ErrInviteInvalid + } + return nil, err + } + return inv, nil +} + +// UpdateRegistrationInvite patches label, max uses, and/or expiry for an editable invite. +// When setExpires is true, expiresAt is applied (nil clears expiry). +func (s *Store) UpdateRegistrationInvite( + ctx context.Context, + id int64, + label *string, + maxUses *int, + setExpires bool, + expiresAt *time.Time, +) (*models.RegistrationInvite, error) { + existing, err := s.GetRegistrationInviteByID(ctx, id) + if err != nil { + return nil, err + } + if existing.Status != models.InviteActive && existing.Status != models.InvitePaused { + return existing, ErrInviteInvalid + } + + newLabel := existing.Label + if label != nil { + newLabel = strings.TrimSpace(*label) + if utf8.RuneCountInString(newLabel) > 120 { + newLabel = string([]rune(newLabel)[:120]) + } + } + + newMax := existing.MaxUses + if maxUses != nil { + if *maxUses < 1 || *maxUses > 10_000 { + return nil, ErrInviteMaxUsesRange + } + if *maxUses < existing.UsedCount { + return nil, ErrInviteMaxUsesTooLow + } + newMax = *maxUses + } + + newExpires := existing.ExpiresAt + if setExpires { + newExpires = expiresAt + } + + row := s.pool.QueryRow(ctx, ` + UPDATE registration_invites + SET label = $2, + max_uses = $3, + expires_at = $4 + WHERE id = $1 AND status IN ('active', 'paused') + RETURNING `+inviteCols, + id, newLabel, newMax, newExpires) + return scanInvite(row) +} + +// EvaluateInviteAvailability checks whether an invite can still accept registrations. +func EvaluateInviteAvailability(inv *models.RegistrationInvite, now time.Time) error { + if inv == nil { + return ErrInviteNotFound + } + switch inv.Status { + case models.InviteRevoked: + return ErrInviteRevoked + case models.InviteExhausted: + return ErrInviteExhausted + case models.InvitePaused: + return ErrInvitePaused + case models.InviteActive: + // ok + default: + return ErrInviteInvalid + } + if inv.ExpiresAt != nil && !inv.ExpiresAt.After(now) { + return ErrInviteExpired + } + if inv.UsedCount >= inv.MaxUses { + return ErrInviteExhausted + } + return nil +} + +// RegisterInviteStudent creates a student, consumes the IP, and atomically consumes one invite seat. +func (s *Store) RegisterInviteStudent( + ctx context.Context, + inviteID int64, + name, email, username, passwordHash, phone, securityQuestion, securityAnswerHash, ip string, +) (*models.User, error) { + ip = strings.TrimSpace(ip) + if ip == "" { + return nil, errors.New("empty registration ip") + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + if blocked, err := isPhoneBlacklistedTx(ctx, tx, phone); err != nil { + return nil, err + } else if blocked { + return nil, ErrPhoneBlacklisted + } + + var taken bool + if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM registration_ips WHERE ip=$1)`, ip).Scan(&taken); err != nil { + return nil, err + } + if taken { + return nil, ErrRegistrationIPTaken + } + + tag, err := tx.Exec(ctx, ` + UPDATE registration_invites + SET used_count = used_count + 1, + status = CASE + WHEN used_count + 1 >= max_uses THEN 'exhausted' + ELSE status + END + WHERE id = $1 + AND status = 'active' + AND used_count < max_uses + AND (expires_at IS NULL OR expires_at > now())`, inviteID) + if err != nil { + return nil, err + } + if tag.RowsAffected() == 0 { + var status string + var used, maxUses int + var expiresAt *time.Time + err := tx.QueryRow(ctx, ` + SELECT status, used_count, max_uses, expires_at + FROM registration_invites WHERE id=$1`, inviteID).Scan(&status, &used, &maxUses, &expiresAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrInviteNotFound + } + if err != nil { + return nil, err + } + switch models.InviteStatus(status) { + case models.InviteRevoked: + return nil, ErrInviteRevoked + case models.InviteExhausted: + return nil, ErrInviteExhausted + case models.InvitePaused: + return nil, ErrInvitePaused + } + if expiresAt != nil && !expiresAt.After(time.Now()) { + return nil, ErrInviteExpired + } + if used >= maxUses { + return nil, ErrInviteExhausted + } + return nil, ErrInviteInvalid + } + + row := tx.QueryRow(ctx, ` + INSERT INTO users (name, email, username, password_hash, role, phone, security_question, security_answer_hash) + VALUES ($1, $2, $3, $4, 'student', $5, $6, $7) + RETURNING `+userCols, + name, email, username, passwordHash, phone, securityQuestion, securityAnswerHash) + u, err := scanUser(row) + if err != nil { + return nil, err + } + // Consume whitelist seat if this phone was listed (invite may use non-whitelist phones). + if err := consumeWhitelistPhoneTx(ctx, tx, phone, u.ID, false); err != nil { + return nil, err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO notification_prefs (user_id) VALUES ($1) + ON CONFLICT (user_id) DO NOTHING`, u.ID); err != nil { + return nil, err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO registration_ips (ip, user_id) VALUES ($1, $2)`, ip, u.ID); err != nil { + if isUniqueViolation(err) { + return nil, ErrRegistrationIPTaken + } + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return u, nil +} diff --git a/backend/internal/store/learning.go b/backend/internal/store/learning.go new file mode 100644 index 0000000..bd710eb --- /dev/null +++ b/backend/internal/store/learning.go @@ -0,0 +1,314 @@ +package store + +import ( + "context" + "time" +) + +// LearningOverview is cohort-level learning analytics for staff. +type LearningOverview struct { + StudentsActive int64 `json:"studentsActive"` + StudentsLocked int64 `json:"studentsLocked"` + StudentsTotal int64 `json:"studentsTotal"` + LessonsTotal int64 `json:"lessonsTotal"` + Completions int64 `json:"completions"` + Attempts7d int64 `json:"attempts7d"` + Passes7d int64 `json:"passes7d"` + PassRate7d float64 `json:"passRate7d"` + AvgWatchPct float64 `json:"avgWatchPct"` + XPAwarded7d int64 `json:"xpAwarded7d"` + StudentsStreaking int64 `json:"studentsStreaking"` +} + +func (s *Store) LearningOverview(ctx context.Context) (*LearningOverview, error) { + o := &LearningOverview{} + if err := s.pool.QueryRow(ctx, ` + SELECT + COUNT(*) FILTER (WHERE role='student'), + COUNT(*) FILTER (WHERE role='student' AND is_active), + COUNT(*) FILTER (WHERE role='student' AND is_locked) + FROM users`).Scan(&o.StudentsTotal, &o.StudentsActive, &o.StudentsLocked); err != nil { + return nil, err + } + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM lessons WHERE is_active`).Scan(&o.LessonsTotal); err != nil { + return nil, err + } + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM lesson_progress WHERE passed_quiz`).Scan(&o.Completions); err != nil { + return nil, err + } + if err := s.pool.QueryRow(ctx, ` + SELECT + COUNT(*), + COUNT(*) FILTER (WHERE status='passed') + FROM quiz_attempts + WHERE created_at >= now() - interval '7 days'`).Scan(&o.Attempts7d, &o.Passes7d); err != nil { + return nil, err + } + if o.Attempts7d > 0 { + o.PassRate7d = float64(o.Passes7d) / float64(o.Attempts7d) * 100 + } + _ = s.pool.QueryRow(ctx, ` + SELECT COALESCE(AVG(watched_pct), 0) + FROM lesson_progress + WHERE watched_pct > 0 AND NOT passed_quiz`).Scan(&o.AvgWatchPct) + _ = s.pool.QueryRow(ctx, ` + SELECT COALESCE(SUM(amount), 0) + FROM xp_events + WHERE created_at >= now() - interval '7 days'`).Scan(&o.XPAwarded7d) + _ = s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM users + WHERE role='student' AND streak_current > 0 AND is_active`).Scan(&o.StudentsStreaking) + return o, nil +} + +// StudentLearningRow is a list row for the learning dashboard. +type StudentLearningRow struct { + ID int64 `json:"id"` + Name string `json:"name"` + Username string `json:"username"` + Email string `json:"email"` + XP int `json:"xp"` + Hearts int `json:"hearts"` + StreakCurrent int `json:"streakCurrent"` + IsLocked bool `json:"isLocked"` + IsActive bool `json:"isActive"` + LessonsPassed int64 `json:"lessonsPassed"` + LessonsTotal int64 `json:"lessonsTotal"` + ProgressPct float64 `json:"progressPct"` + Attempts7d int64 `json:"attempts7d"` + Passes7d int64 `json:"passes7d"` + PassRate7d float64 `json:"passRate7d"` + LastActivityAt *time.Time `json:"lastActivityAt,omitempty"` +} + +func (s *Store) ListStudentLearning(ctx context.Context, q string) ([]StudentLearningRow, error) { + var lessonsTotal int64 + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM lessons WHERE is_active`).Scan(&lessonsTotal); err != nil { + return nil, err + } + rows, err := s.pool.Query(ctx, ` + SELECT u.id, u.name, u.username, u.email, u.xp, u.hearts, u.streak_current, + u.is_locked, u.is_active, u.last_activity_date, + COALESCE(p.passed, 0), + COALESCE(a.attempts_7d, 0), + COALESCE(a.passes_7d, 0) + FROM users u + LEFT JOIN ( + SELECT user_id, COUNT(*) FILTER (WHERE passed_quiz) AS passed + FROM lesson_progress GROUP BY user_id + ) p ON p.user_id = u.id + LEFT JOIN ( + SELECT user_id, + COUNT(*) AS attempts_7d, + COUNT(*) FILTER (WHERE status='passed') AS passes_7d + FROM quiz_attempts + WHERE created_at >= now() - interval '7 days' + GROUP BY user_id + ) a ON a.user_id = u.id + WHERE u.role = 'student' + AND ($1 = '' OR u.name ILIKE '%'||$1||'%' OR u.username ILIKE '%'||$1||'%' OR u.email ILIKE '%'||$1||'%') + ORDER BY u.xp DESC, u.id ASC + LIMIT 500`, q) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]StudentLearningRow, 0) + for rows.Next() { + var r StudentLearningRow + r.LessonsTotal = lessonsTotal + if err := rows.Scan( + &r.ID, &r.Name, &r.Username, &r.Email, &r.XP, &r.Hearts, &r.StreakCurrent, + &r.IsLocked, &r.IsActive, &r.LastActivityAt, + &r.LessonsPassed, &r.Attempts7d, &r.Passes7d, + ); err != nil { + return nil, err + } + if r.LessonsTotal > 0 { + r.ProgressPct = float64(r.LessonsPassed) / float64(r.LessonsTotal) * 100 + } + if r.Attempts7d > 0 { + r.PassRate7d = float64(r.Passes7d) / float64(r.Attempts7d) * 100 + } + out = append(out, r) + } + return out, rows.Err() +} + +type LessonLearningDetail struct { + LessonID int64 `json:"lessonId"` + ChapterID int64 `json:"chapterId"` + ChapterTitle string `json:"chapterTitle"` + Title string `json:"title"` + SortOrder int `json:"sortOrder"` + WatchedPct float64 `json:"watchedPct"` + QuizUnlocked bool `json:"quizUnlocked"` + PassedQuiz bool `json:"passedQuiz"` + QuizCompletedAt *time.Time `json:"quizCompletedAt,omitempty"` + LastAttemptStatus string `json:"lastAttemptStatus,omitempty"` + LastAttemptScore float64 `json:"lastAttemptScore,omitempty"` +} + +type AttemptLearningDetail struct { + ID int64 `json:"id"` + LessonID int64 `json:"lessonId"` + LessonTitle string `json:"lessonTitle"` + CorrectCount int `json:"correctCount"` + Total int `json:"total"` + ScorePct float64 `json:"scorePct"` + Status string `json:"status"` + HeartsLost int `json:"heartsLost"` + CreatedAt time.Time `json:"createdAt"` +} + +type XPEventDetail struct { + ID int64 `json:"id"` + Amount int `json:"amount"` + Source string `json:"source"` + RefID int64 `json:"refId"` + CreatedAt time.Time `json:"createdAt"` +} + +type WatchSessionDetail struct { + LessonID int64 `json:"lessonId"` + LessonTitle string `json:"lessonTitle"` + StartedAt time.Time `json:"startedAt"` + LastHeartbeatAt time.Time `json:"lastHeartbeatAt"` + LastPosition float64 `json:"lastPosition"` +} + +type UserLearningDetail struct { + User StudentLearningRow `json:"user"` + Lessons []LessonLearningDetail `json:"lessons"` + Attempts []AttemptLearningDetail `json:"attempts"` + XPEvents []XPEventDetail `json:"xpEvents"` + Sessions []WatchSessionDetail `json:"sessions"` +} + +func (s *Store) UserLearningDetail(ctx context.Context, userID int64) (*UserLearningDetail, error) { + u, err := s.GetUserByID(ctx, userID) + if err != nil { + return nil, err + } + list, err := s.ListStudentLearning(ctx, "") + if err != nil { + return nil, err + } + var row StudentLearningRow + found := false + for _, r := range list { + if r.ID == userID { + row = r + found = true + break + } + } + if !found { + row = StudentLearningRow{ + ID: u.ID, Name: u.Name, Username: u.Username, Email: u.Email, + XP: u.XP, Hearts: u.Hearts, StreakCurrent: u.StreakCurrent, + IsLocked: u.IsLocked, IsActive: u.IsActive, LastActivityAt: u.LastActivityDate, + } + _ = s.pool.QueryRow(ctx, `SELECT count(*) FROM lessons WHERE is_active`).Scan(&row.LessonsTotal) + _ = s.pool.QueryRow(ctx, `SELECT count(*) FROM lesson_progress WHERE user_id=$1 AND passed_quiz`, userID).Scan(&row.LessonsPassed) + if row.LessonsTotal > 0 { + row.ProgressPct = float64(row.LessonsPassed) / float64(row.LessonsTotal) * 100 + } + } + + detail := &UserLearningDetail{User: row, Lessons: []LessonLearningDetail{}, Attempts: []AttemptLearningDetail{}, XPEvents: []XPEventDetail{}, Sessions: []WatchSessionDetail{}} + + lrows, err := s.pool.Query(ctx, ` + SELECT l.id, l.chapter_id, c.title, l.title, l.sort_order, + COALESCE(p.watched_pct, 0), COALESCE(p.quiz_unlocked, false), COALESCE(p.passed_quiz, false), p.quiz_completed_at, + COALESCE(la.status, ''), COALESCE(la.score_pct, 0) + FROM lessons l + JOIN chapters c ON c.id = l.chapter_id + LEFT JOIN lesson_progress p ON p.lesson_id = l.id AND p.user_id = $1 + LEFT JOIN LATERAL ( + SELECT status, score_pct FROM quiz_attempts + WHERE user_id=$1 AND lesson_id=l.id + ORDER BY created_at DESC LIMIT 1 + ) la ON true + WHERE l.is_active + ORDER BY c.sort_order, l.sort_order, l.id`, userID) + if err != nil { + return nil, err + } + defer lrows.Close() + for lrows.Next() { + var d LessonLearningDetail + if err := lrows.Scan( + &d.LessonID, &d.ChapterID, &d.ChapterTitle, &d.Title, &d.SortOrder, + &d.WatchedPct, &d.QuizUnlocked, &d.PassedQuiz, &d.QuizCompletedAt, + &d.LastAttemptStatus, &d.LastAttemptScore, + ); err != nil { + return nil, err + } + detail.Lessons = append(detail.Lessons, d) + } + if err := lrows.Err(); err != nil { + return nil, err + } + + arows, err := s.pool.Query(ctx, ` + SELECT a.id, a.lesson_id, l.title, a.correct_count, a.total, a.score_pct, a.status, a.hearts_lost, a.created_at + FROM quiz_attempts a + JOIN lessons l ON l.id = a.lesson_id + WHERE a.user_id = $1 + ORDER BY a.created_at DESC + LIMIT 50`, userID) + if err != nil { + return nil, err + } + defer arows.Close() + for arows.Next() { + var a AttemptLearningDetail + if err := arows.Scan(&a.ID, &a.LessonID, &a.LessonTitle, &a.CorrectCount, &a.Total, &a.ScorePct, &a.Status, &a.HeartsLost, &a.CreatedAt); err != nil { + return nil, err + } + detail.Attempts = append(detail.Attempts, a) + } + if err := arows.Err(); err != nil { + return nil, err + } + + xrows, err := s.pool.Query(ctx, ` + SELECT id, amount, source, COALESCE(ref_id, 0), created_at + FROM xp_events WHERE user_id=$1 + ORDER BY created_at DESC LIMIT 40`, userID) + if err != nil { + return nil, err + } + defer xrows.Close() + for xrows.Next() { + var x XPEventDetail + if err := xrows.Scan(&x.ID, &x.Amount, &x.Source, &x.RefID, &x.CreatedAt); err != nil { + return nil, err + } + detail.XPEvents = append(detail.XPEvents, x) + } + if err := xrows.Err(); err != nil { + return nil, err + } + + srows, err := s.pool.Query(ctx, ` + SELECT ws.lesson_id, l.title, ws.started_at, ws.last_heartbeat_at, ws.last_position + FROM watch_sessions ws + JOIN lessons l ON l.id = ws.lesson_id + WHERE ws.user_id = $1 + ORDER BY ws.last_heartbeat_at DESC + LIMIT 20`, userID) + if err != nil { + return nil, err + } + defer srows.Close() + for srows.Next() { + var w WatchSessionDetail + if err := srows.Scan(&w.LessonID, &w.LessonTitle, &w.StartedAt, &w.LastHeartbeatAt, &w.LastPosition); err != nil { + return nil, err + } + detail.Sessions = append(detail.Sessions, w) + } + return detail, srows.Err() +} diff --git a/backend/internal/store/learning_test.go b/backend/internal/store/learning_test.go new file mode 100644 index 0000000..4e100b1 --- /dev/null +++ b/backend/internal/store/learning_test.go @@ -0,0 +1,36 @@ +package store_test + +import ( + "context" + "os" + "testing" + + "pargar/backend/internal/db" + "pargar/backend/internal/store" +) + +func TestLearningOverviewSmoke(t *testing.T) { + dsn := os.Getenv("DATABASE_URL") + if dsn == "" { + dsn = "postgres://pargar:f479ad70f076f25594b73b767691be4b295901c4430ed208@localhost:5432/pargar?sslmode=disable" + } + ctx := context.Background() + pool, err := db.Connect(ctx, dsn) + if err != nil { + t.Skip("postgres unavailable:", err) + } + defer pool.Close() + st := store.New(pool) + o, err := st.LearningOverview(ctx) + if err != nil { + t.Fatal(err) + } + if o == nil { + t.Fatal("nil overview") + } + students, err := st.ListStudentLearning(ctx, "") + if err != nil { + t.Fatal(err) + } + _ = students +} diff --git a/backend/internal/store/store.go b/backend/internal/store/store.go new file mode 100644 index 0000000..7161d8b --- /dev/null +++ b/backend/internal/store/store.go @@ -0,0 +1,1764 @@ +package store + +import ( + "context" + "encoding/json" + "errors" + "strconv" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + + "pargar/backend/internal/models" +) + +var ErrNotFound = errors.New("not found") +var ErrPinLimit = errors.New("pin limit") + +const MaxHearts = 3 + +type Store struct { + pool *pgxpool.Pool +} + +func New(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} + +func (s *Store) Pool() *pgxpool.Pool { return s.pool } + +type Tx struct { + queries pgx.Tx +} + +func (s *Store) Begin(ctx context.Context) (*Tx, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + return &Tx{queries: tx}, nil +} + +func (t *Tx) Commit(ctx context.Context) error { return t.queries.Commit(ctx) } +func (t *Tx) Rollback(ctx context.Context) error { + return t.queries.Rollback(ctx) +} + +// ---- Users ---- + +func (s *Store) CreateUser(ctx context.Context, name, email, username, passwordHash, phone, securityQuestion, securityAnswerHash string, role models.Role) (*models.User, error) { + row := s.pool.QueryRow(ctx, ` + INSERT INTO users (name, email, username, password_hash, role, phone, security_question, security_answer_hash) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + RETURNING `+userCols, + name, email, username, passwordHash, role.String(), phone, securityQuestion, securityAnswerHash) + u, err := scanUser(row) + if err != nil { + return nil, err + } + if _, err := s.pool.Exec(ctx, ` + INSERT INTO notification_prefs (user_id) VALUES ($1) + ON CONFLICT (user_id) DO NOTHING`, u.ID); err != nil { + return nil, err + } + return u, nil +} + +type rowScanner interface { + Scan(dest ...any) error +} + +var _ rowScanner = pgx.Row(nil) + +func scanUser(row rowScanner) (*models.User, error) { + var u models.User + err := row.Scan(&u.ID, &u.Name, &u.Email, &u.Username, &u.PasswordHash, &u.Role, + &u.XP, &u.Hearts, &u.HeartsUpdatedAt, &u.StreakCurrent, &u.StreakLongest, + &u.LastActivityDate, &u.IsLocked, &u.LockedAt, &u.UnlockedBy, &u.IsActive, + &u.AvatarVariant, &u.AvatarPalette, &u.AvatarPhoto, + &u.Phone, &u.TelegramID, &u.TelegramUsername, &u.SecurityQuestion, &u.SecurityAnswerHash, + &u.CreatedAt) + if err != nil { + return nil, err + } + u.HasSecurityAnswer = u.SecurityAnswerHash != "" + return &u, nil +} + +const userCols = `id, name, email, username, password_hash, role, xp, hearts, hearts_updated_at, + streak_current, streak_longest, last_activity_date, is_locked, locked_at, + unlocked_by, is_active, avatar_variant, avatar_palette, avatar_photo, + phone, telegram_id, telegram_username, security_question, security_answer_hash, created_at` + +func (s *Store) GetUserByEmail(ctx context.Context, email string) (*models.User, error) { + row := s.pool.QueryRow(ctx, `SELECT `+userCols+` FROM users WHERE email=$1`, email) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) GetUserByUsername(ctx context.Context, username string) (*models.User, error) { + row := s.pool.QueryRow(ctx, `SELECT `+userCols+` FROM users WHERE lower(username)=$1`, username) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +// GetUserByLogin resolves an email or username identifier. +func (s *Store) GetUserByLogin(ctx context.Context, login string) (*models.User, error) { + login = strings.TrimSpace(login) + if strings.Contains(login, "@") { + return s.GetUserByEmail(ctx, strings.ToLower(login)) + } + return s.GetUserByUsername(ctx, strings.ToLower(login)) +} + +func (s *Store) GetUserByID(ctx context.Context, id int64) (*models.User, error) { + row := s.pool.QueryRow(ctx, `SELECT `+userCols+` FROM users WHERE id=$1`, id) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) UpdateUserStats(ctx context.Context, userID int64, xp int, streak int, streakLongest int, lastActivity *time.Time) error { + _, err := s.pool.Exec(ctx, ` + UPDATE users SET xp=$1, streak_current=$2, streak_longest=$3, last_activity_date=$4 + WHERE id=$5`, xp, streak, streakLongest, lastActivity, userID) + return err +} + +// UpdateStreakOnly updates streak fields without rewriting XP (avoids clobbering concurrent AddUserXP). +func (s *Store) UpdateStreakOnly(ctx context.Context, userID int64, streak int, streakLongest int, lastActivity *time.Time) error { + _, err := s.pool.Exec(ctx, ` + UPDATE users SET streak_current=$1, streak_longest=$2, last_activity_date=$3 + WHERE id=$4`, streak, streakLongest, lastActivity, userID) + return err +} + +func (s *Store) AddUserXP(ctx context.Context, userID int64, amount int) (*models.User, error) { + row := s.pool.QueryRow(ctx, `UPDATE users SET xp = xp + $1 WHERE id=$2 RETURNING `+userCols, amount, userID) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) SetHearts(ctx context.Context, userID int64, hearts int) error { + violation := hearts > MaxHearts + if hearts < 0 { + hearts = 0 + } + if violation { + hearts = MaxHearts + } + _, err := s.pool.Exec(ctx, ` + UPDATE users + SET hearts=$1, hearts_updated_at=now(), is_locked=is_locked OR $3, + locked_at=CASE WHEN $3 THEN COALESCE(locked_at, now()) ELSE locked_at END + WHERE id=$2`, hearts, userID, violation) + return err +} + +func (s *Store) UpdateAvatar(ctx context.Context, userID int64, variant, palette *string, photo *string) (*models.User, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE users SET + avatar_variant = COALESCE($2::text, avatar_variant), + avatar_palette = COALESCE($3::text, avatar_palette), + avatar_photo = COALESCE($4::text, avatar_photo) + WHERE id=$1 RETURNING `+userCols, userID, variant, palette, photo) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) LockUser(ctx context.Context, userID int64) (*models.User, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE users SET is_locked=true, locked_at=now() WHERE id=$1 RETURNING `+userCols, userID) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) UnlockUser(ctx context.Context, userID int64, by int64) (*models.User, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE users SET is_locked=false, locked_at=NULL, unlocked_by=$1, hearts=3, hearts_updated_at=now() + WHERE id=$2 RETURNING `+userCols, by, userID) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) ListUsers(ctx context.Context, query string, role string, limit, offset int) ([]models.User, int64, error) { + if limit <= 0 { + limit = 10 + } + if limit > 100 { + limit = 100 + } + if offset < 0 { + offset = 0 + } + where := `($1='' OR name ILIKE '%'||$1||'%' OR email ILIKE '%'||$1||'%' OR username ILIKE '%'||$1||'%')` + args := []any{query} + if role != "" { + args = append(args, role) + where += ` AND role=$2` + } + var total int64 + countQ := `SELECT COUNT(*) FROM users WHERE ` + where + if err := s.pool.QueryRow(ctx, countQ, args...).Scan(&total); err != nil { + return nil, 0, err + } + limitIdx := len(args) + 1 + offsetIdx := len(args) + 2 + args = append(args, limit, offset) + q := `SELECT ` + userCols + ` FROM users WHERE ` + where + + ` ORDER BY created_at ASC LIMIT $` + strconv.Itoa(limitIdx) + ` OFFSET $` + strconv.Itoa(offsetIdx) + rows, err := s.pool.Query(ctx, q, args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + var out []models.User + for rows.Next() { + u, err := scanUser(rows) + if err != nil { + return nil, 0, err + } + out = append(out, *u) + } + return out, total, rows.Err() +} + +func (s *Store) ListMentors(ctx context.Context) ([]models.User, error) { + rows, err := s.pool.Query(ctx, `SELECT `+userCols+` FROM users WHERE role IN ('mentor','admin') AND is_active ORDER BY name`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.User + for rows.Next() { + u, err := scanUser(rows) + if err != nil { + return nil, err + } + out = append(out, *u) + } + return out, rows.Err() +} + +func (s *Store) UpdateUser(ctx context.Context, id int64, name, email, username, passwordHash, phone, securityQuestion, securityAnswerHash string, role models.Role, hearts, xp int) (*models.User, error) { + violation := hearts > MaxHearts + if hearts < 0 { + hearts = 0 + } + if violation { + hearts = MaxHearts + } + row := s.pool.QueryRow(ctx, ` + UPDATE users SET + name = COALESCE(NULLIF($2::text, ''), name), + email = COALESCE(NULLIF($3::text, ''), email), + username = COALESCE(NULLIF($4::text, ''), username), + password_hash = CASE WHEN $5::text = '' THEN password_hash ELSE $5 END, + role = $6::user_role, + hearts = $7, + xp = $8, + is_locked = is_locked OR $9, + locked_at = CASE WHEN $9 THEN COALESCE(locked_at, now()) ELSE locked_at END, + phone = COALESCE($10, phone), + security_question = CASE WHEN $11::text = '' THEN security_question ELSE $11 END, + security_answer_hash = CASE WHEN $12::text = '' THEN security_answer_hash ELSE $12 END + WHERE id=$1 RETURNING `+userCols, + id, name, email, username, passwordHash, role.String(), hearts, xp, violation, phone, securityQuestion, securityAnswerHash) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) GetUserByPhone(ctx context.Context, phone string) (*models.User, error) { + phone = strings.TrimSpace(phone) + if phone == "" { + return nil, ErrNotFound + } + row := s.pool.QueryRow(ctx, `SELECT `+userCols+` FROM users WHERE phone=$1`, phone) + u, err := scanUser(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return u, err +} + +func (s *Store) SetUserPassword(ctx context.Context, id int64, passwordHash string) error { + tag, err := s.pool.Exec(ctx, `UPDATE users SET password_hash=$1 WHERE id=$2`, passwordHash, id) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +// ErrRegistrationIPTaken is returned when an IP has already completed public registration. +var ErrRegistrationIPTaken = errors.New("registration ip taken") + +func (s *Store) HasRegistrationIP(ctx context.Context, ip string) (bool, error) { + ip = strings.TrimSpace(ip) + if ip == "" { + return false, nil + } + var exists bool + err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM registration_ips WHERE ip=$1)`, ip).Scan(&exists) + return exists, err +} + +// RegisterPublicStudent creates a student and consumes the IP atomically. +// When requireWhitelist is true, the phone must be an available whitelist entry. +// Concurrent same-IP registrations: one commits; the other rolls back on unique IP conflict. +func (s *Store) RegisterPublicStudent( + ctx context.Context, + name, email, username, passwordHash, phone, securityQuestion, securityAnswerHash, ip string, + requireWhitelist bool, +) (*models.User, error) { + ip = strings.TrimSpace(ip) + if ip == "" { + return nil, errors.New("empty registration ip") + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + if blocked, err := isPhoneBlacklistedTx(ctx, tx, phone); err != nil { + return nil, err + } else if blocked { + return nil, ErrPhoneBlacklisted + } + + var taken bool + if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM registration_ips WHERE ip=$1)`, ip).Scan(&taken); err != nil { + return nil, err + } + if taken { + return nil, ErrRegistrationIPTaken + } + + row := tx.QueryRow(ctx, ` + INSERT INTO users (name, email, username, password_hash, role, phone, security_question, security_answer_hash) + VALUES ($1, $2, $3, $4, 'student', $5, $6, $7) + RETURNING `+userCols, + name, email, username, passwordHash, phone, securityQuestion, securityAnswerHash) + u, err := scanUser(row) + if err != nil { + return nil, err + } + if err := consumeWhitelistPhoneTx(ctx, tx, phone, u.ID, requireWhitelist); err != nil { + return nil, err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO notification_prefs (user_id) VALUES ($1) + ON CONFLICT (user_id) DO NOTHING`, u.ID); err != nil { + return nil, err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO registration_ips (ip, user_id) VALUES ($1, $2)`, ip, u.ID); err != nil { + if isUniqueViolation(err) { + return nil, ErrRegistrationIPTaken + } + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return u, nil +} + +func isUniqueViolation(err error) bool { + var pe *pgconn.PgError + return errors.As(err, &pe) && pe.Code == "23505" +} + +// IsUniqueViolation reports a Postgres unique_violation (23505). +func IsUniqueViolation(err error) bool { + return isUniqueViolation(err) +} + +func (s *Store) CountActiveLessons(ctx context.Context) (int, error) { + var n int + err := s.pool.QueryRow(ctx, `SELECT COUNT(*) FROM lessons WHERE is_active`).Scan(&n) + return n, err +} + +func (s *Store) CountPassedActiveLessons(ctx context.Context, userID int64) (int, error) { + var n int + err := s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM lesson_progress lp + JOIN lessons l ON l.id = lp.lesson_id + WHERE lp.user_id=$1 AND lp.passed_quiz AND l.is_active`, userID).Scan(&n) + return n, err +} + +func (s *Store) HasCompletedCurriculum(ctx context.Context, userID int64) (bool, error) { + active, err := s.CountActiveLessons(ctx) + if err != nil { + return false, err + } + if active == 0 { + return false, nil + } + passed, err := s.CountPassedActiveLessons(ctx, userID) + if err != nil { + return false, err + } + return passed >= active, nil +} + +func (s *Store) GetCertificateByUser(ctx context.Context, userID int64) (*models.Certificate, error) { + row := s.pool.QueryRow(ctx, ` + SELECT id, user_id, public_id, full_name, issued_at, revoked_at + FROM certificates WHERE user_id=$1`, userID) + return scanCertificate(row) +} + +func (s *Store) GetCertificateByPublicID(ctx context.Context, publicID string) (*models.Certificate, error) { + row := s.pool.QueryRow(ctx, ` + SELECT id, user_id, public_id, full_name, issued_at, revoked_at + FROM certificates WHERE public_id=$1`, publicID) + return scanCertificate(row) +} + +func scanCertificate(row rowScanner) (*models.Certificate, error) { + var c models.Certificate + err := row.Scan(&c.ID, &c.UserID, &c.PublicID, &c.FullName, &c.IssuedAt, &c.RevokedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return &c, nil +} + +func (s *Store) CreateCertificate(ctx context.Context, userID int64, publicID, fullName string) (*models.Certificate, error) { + row := s.pool.QueryRow(ctx, ` + INSERT INTO certificates (user_id, public_id, full_name) + VALUES ($1, $2, $3) + ON CONFLICT (user_id) DO UPDATE SET + full_name = EXCLUDED.full_name, + revoked_at = NULL + RETURNING id, user_id, public_id, full_name, issued_at, revoked_at`, + userID, publicID, fullName) + return scanCertificate(row) +} + +const physicalOrderCols = `id, certificate_id, user_id, status, note, + recipient_name, phone, address, city, postal_code, tracking_code, + window_ends_at, created_at, updated_at` + +func (s *Store) CreatePhysicalOrder(ctx context.Context, certID, userID int64, windowEndsAt time.Time, ship models.CertificatePhysicalOrder) (*models.CertificatePhysicalOrder, error) { + row := s.pool.QueryRow(ctx, ` + INSERT INTO certificate_physical_orders ( + certificate_id, user_id, window_ends_at, + recipient_name, phone, address, city, postal_code + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8) + RETURNING `+physicalOrderCols, + certID, userID, windowEndsAt, + ship.RecipientName, ship.Phone, ship.Address, ship.City, ship.PostalCode) + return scanPhysicalOrder(row) +} + +func (s *Store) GetLatestPhysicalOrder(ctx context.Context, userID int64) (*models.CertificatePhysicalOrder, error) { + row := s.pool.QueryRow(ctx, ` + SELECT `+physicalOrderCols+` + FROM certificate_physical_orders + WHERE user_id=$1 + ORDER BY created_at DESC LIMIT 1`, userID) + return scanPhysicalOrder(row) +} + +func (s *Store) GetPhysicalOrder(ctx context.Context, id int64) (*models.CertificatePhysicalOrder, error) { + row := s.pool.QueryRow(ctx, ` + SELECT `+physicalOrderCols+` + FROM certificate_physical_orders WHERE id=$1`, id) + return scanPhysicalOrder(row) +} + +func (s *Store) ListPhysicalOrders(ctx context.Context, status string) ([]models.CertificatePhysicalOrder, error) { + q := ` + SELECT o.id, o.certificate_id, o.user_id, o.status, o.note, + o.recipient_name, o.phone, o.address, o.city, o.postal_code, o.tracking_code, + o.window_ends_at, o.created_at, o.updated_at, + u.name, c.public_id + FROM certificate_physical_orders o + JOIN users u ON u.id = o.user_id + JOIN certificates c ON c.id = o.certificate_id` + args := []any{} + if status != "" { + q += ` WHERE o.status=$1` + args = append(args, status) + } + q += ` ORDER BY o.created_at DESC LIMIT 200` + rows, err := s.pool.Query(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]models.CertificatePhysicalOrder, 0) + for rows.Next() { + var o models.CertificatePhysicalOrder + if err := rows.Scan(&o.ID, &o.CertificateID, &o.UserID, &o.Status, &o.Note, + &o.RecipientName, &o.Phone, &o.Address, &o.City, &o.PostalCode, &o.TrackingCode, + &o.WindowEndsAt, &o.CreatedAt, &o.UpdatedAt, &o.UserName, &o.PublicID); err != nil { + return nil, err + } + out = append(out, o) + } + return out, rows.Err() +} + +func (s *Store) UpdatePhysicalOrder(ctx context.Context, id int64, status models.PhysicalOrderStatus, note, tracking string) (*models.CertificatePhysicalOrder, error) { + row := s.pool.QueryRow(ctx, ` + UPDATE certificate_physical_orders + SET status=$2, + note=CASE WHEN $3='' THEN note ELSE $3 END, + tracking_code=CASE WHEN $4='' THEN tracking_code ELSE $4 END, + updated_at=now() + WHERE id=$1 + RETURNING `+physicalOrderCols, + id, string(status), note, tracking) + return scanPhysicalOrder(row) +} + +func scanPhysicalOrder(row rowScanner) (*models.CertificatePhysicalOrder, error) { + var o models.CertificatePhysicalOrder + err := row.Scan( + &o.ID, &o.CertificateID, &o.UserID, &o.Status, &o.Note, + &o.RecipientName, &o.Phone, &o.Address, &o.City, &o.PostalCode, &o.TrackingCode, + &o.WindowEndsAt, &o.CreatedAt, &o.UpdatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return &o, nil +} + +func (s *Store) CreateUnlockRequest(ctx context.Context, userID int64, note string) (*models.UnlockRequest, error) { + row := s.pool.QueryRow(ctx, ` + INSERT INTO unlock_requests (user_id, note) VALUES ($1, $2) + RETURNING id, user_id, note, created_at, resolved_at`, userID, note) + var ur models.UnlockRequest + if err := row.Scan(&ur.ID, &ur.UserID, &ur.Note, &ur.CreatedAt, &ur.ResolvedAt); err != nil { + return nil, err + } + return &ur, nil +} + +func (s *Store) LatestOpenUnlockRequest(ctx context.Context, userID int64, within time.Duration) (*models.UnlockRequest, error) { + since := time.Now().Add(-within) + row := s.pool.QueryRow(ctx, ` + SELECT id, user_id, note, created_at, resolved_at + FROM unlock_requests + WHERE user_id=$1 AND resolved_at IS NULL AND created_at > $2 + ORDER BY created_at DESC LIMIT 1`, userID, since) + var ur models.UnlockRequest + err := row.Scan(&ur.ID, &ur.UserID, &ur.Note, &ur.CreatedAt, &ur.ResolvedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return &ur, nil +} + +func (s *Store) ResolveUnlockRequests(ctx context.Context, userID int64) error { + _, err := s.pool.Exec(ctx, ` + UPDATE unlock_requests SET resolved_at=now() + WHERE user_id=$1 AND resolved_at IS NULL`, userID) + return err +} + +func (s *Store) GetSetting(ctx context.Context, key string) (string, error) { + var v string + err := s.pool.QueryRow(ctx, `SELECT value FROM app_settings WHERE key=$1`, key).Scan(&v) + if errors.Is(err, pgx.ErrNoRows) { + return "", ErrNotFound + } + return v, err +} + +func (s *Store) SetSetting(ctx context.Context, key, value string) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO app_settings (key, value) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value`, key, value) + return err +} + +func (s *Store) DeleteUser(ctx context.Context, id int64) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + queries := []string{ + `DELETE FROM watch_heartbeats WHERE user_id=$1 OR session_id IN (SELECT id FROM watch_sessions WHERE user_id=$1)`, + `DELETE FROM watch_sessions WHERE user_id=$1`, + `DELETE FROM quiz_attempts WHERE user_id=$1`, + `DELETE FROM lesson_progress WHERE user_id=$1`, + `DELETE FROM xp_events WHERE user_id=$1`, + `DELETE FROM weekly_leaderboard WHERE user_id=$1`, + `DELETE FROM event_rsvps WHERE user_id=$1`, + `DELETE FROM chat_reactions WHERE user_id=$1`, + `DELETE FROM chat_messages WHERE user_id=$1 OR mentor_id=$1`, + `DELETE FROM notifications WHERE user_id=$1`, + `DELETE FROM notification_prefs WHERE user_id=$1`, + `DELETE FROM refresh_tokens WHERE user_id=$1`, + `DELETE FROM reminders_sent WHERE user_id=$1`, + `UPDATE users SET unlocked_by=NULL WHERE unlocked_by=$1`, + `UPDATE events SET created_by=NULL WHERE created_by=$1`, + `DELETE FROM users WHERE id=$1`, + } + for _, q := range queries { + if _, err := tx.Exec(ctx, q, id); err != nil { + return err + } + } + return tx.Commit(ctx) +} + +// ---- Chapters / Lessons / Questions ---- + +func (s *Store) CreateChapter(ctx context.Context, c *models.Chapter) (int64, error) { + err := s.pool.QueryRow(ctx, ` + INSERT INTO chapters (title, description, icon, sort_order) VALUES ($1,$2,$3,$4) RETURNING id`, + c.Title, c.Description, c.Icon, c.SortOrder).Scan(&c.ID) + return c.ID, err +} + +func (s *Store) UpdateChapter(ctx context.Context, id int64, c *models.Chapter) error { + _, err := s.pool.Exec(ctx, ` + UPDATE chapters SET title=$1, description=$2, icon=$3, sort_order=$4 WHERE id=$5`, + c.Title, c.Description, c.Icon, c.SortOrder, id) + return err +} + +func (s *Store) DeleteChapter(ctx context.Context, id int64) error { + _, err := s.pool.Exec(ctx, `DELETE FROM chapters WHERE id=$1`, id) + return err +} + +func (s *Store) ListChapters(ctx context.Context) ([]models.Chapter, error) { + rows, err := s.pool.Query(ctx, `SELECT id, title, description, icon, sort_order FROM chapters ORDER BY sort_order, id`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.Chapter + for rows.Next() { + var c models.Chapter + if err := rows.Scan(&c.ID, &c.Title, &c.Description, &c.Icon, &c.SortOrder); err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + +func (s *Store) CreateLesson(ctx context.Context, l *models.Lesson) (int64, error) { + err := s.pool.QueryRow(ctx, ` + INSERT INTO lessons (chapter_id, title, description, video_key, duration_seconds, + completion_threshold_pct, requires_lesson_id, x, y, sort_order, xp_reward, is_active) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) RETURNING id`, + l.ChapterID, l.Title, l.Description, l.VideoKey, l.DurationSeconds, + l.CompletionThresholdPct, l.RequiresLessonID, l.X, l.Y, l.SortOrder, l.XPReward, l.IsActive).Scan(&l.ID) + return l.ID, err +} + +func (s *Store) UpdateLesson(ctx context.Context, id int64, l *models.Lesson) error { + _, err := s.pool.Exec(ctx, ` + UPDATE lessons SET chapter_id=$1, title=$2, description=$3, video_key=$4, duration_seconds=$5, + completion_threshold_pct=$6, requires_lesson_id=$7, x=$8, y=$9, sort_order=$10, xp_reward=$11, is_active=$12 + WHERE id=$13`, + l.ChapterID, l.Title, l.Description, l.VideoKey, l.DurationSeconds, + l.CompletionThresholdPct, l.RequiresLessonID, l.X, l.Y, l.SortOrder, l.XPReward, l.IsActive, id) + return err +} + +func (s *Store) DeleteLesson(ctx context.Context, id int64) error { + _, err := s.pool.Exec(ctx, `DELETE FROM lessons WHERE id=$1`, id) + return err +} + +func (s *Store) GetLesson(ctx context.Context, id int64) (*models.Lesson, error) { + row := s.pool.QueryRow(ctx, ` + SELECT id, chapter_id, title, description, video_key, duration_seconds, + completion_threshold_pct, requires_lesson_id, x, y, sort_order, xp_reward, is_active + FROM lessons WHERE id=$1`, id) + var l models.Lesson + err := row.Scan(&l.ID, &l.ChapterID, &l.Title, &l.Description, &l.VideoKey, &l.DurationSeconds, + &l.CompletionThresholdPct, &l.RequiresLessonID, &l.X, &l.Y, &l.SortOrder, &l.XPReward, &l.IsActive) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return &l, err +} + +func (s *Store) ListLessons(ctx context.Context, includeInactive bool) ([]models.Lesson, error) { + q := `SELECT id, chapter_id, title, description, video_key, duration_seconds, + completion_threshold_pct, requires_lesson_id, x, y, sort_order, xp_reward, is_active FROM lessons` + if !includeInactive { + q += ` WHERE is_active` + } + q += ` ORDER BY chapter_id, sort_order, id` + rows, err := s.pool.Query(ctx, q) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.Lesson + for rows.Next() { + var l models.Lesson + if err := rows.Scan(&l.ID, &l.ChapterID, &l.Title, &l.Description, &l.VideoKey, &l.DurationSeconds, + &l.CompletionThresholdPct, &l.RequiresLessonID, &l.X, &l.Y, &l.SortOrder, &l.XPReward, &l.IsActive); err != nil { + return nil, err + } + out = append(out, l) + } + return out, rows.Err() +} + +func (s *Store) CreateQuestion(ctx context.Context, q *models.MCQQuestion) (int64, error) { + opts, _ := json.Marshal(q.Options) + err := s.pool.QueryRow(ctx, ` + INSERT INTO mcq_questions (lesson_id, position, question, options, answer_index, explanation) + VALUES ($1,$2,$3,$4,$5,$6) RETURNING id`, + q.LessonID, q.Position, q.Question, opts, q.AnswerIndex, q.Explanation).Scan(&q.ID) + return q.ID, err +} + +func (s *Store) UpdateQuestion(ctx context.Context, id int64, q *models.MCQQuestion) error { + opts, _ := json.Marshal(q.Options) + _, err := s.pool.Exec(ctx, ` + UPDATE mcq_questions SET position=$1, question=$2, options=$3, answer_index=$4, explanation=$5 WHERE id=$6`, + q.Position, q.Question, opts, q.AnswerIndex, q.Explanation, id) + return err +} + +func (s *Store) DeleteQuestion(ctx context.Context, id int64) error { + _, err := s.pool.Exec(ctx, `DELETE FROM mcq_questions WHERE id=$1`, id) + return err +} + +func (s *Store) ListQuestions(ctx context.Context, lessonID int64, withAnswers bool) ([]models.MCQQuestion, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, lesson_id, position, question, options, answer_index, explanation + FROM mcq_questions WHERE lesson_id=$1 ORDER BY position, id`, lessonID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.MCQQuestion + for rows.Next() { + var q models.MCQQuestion + var opts []byte + if err := rows.Scan(&q.ID, &q.LessonID, &q.Position, &q.Question, &opts, &q.AnswerIndex, &q.Explanation); err != nil { + return nil, err + } + _ = json.Unmarshal(opts, &q.Options) + if !withAnswers { + q.AnswerIndex = -1 + } + out = append(out, q) + } + return out, rows.Err() +} + +// ---- Watch sessions / heartbeats / progress ---- + +func (s *Store) GetOrCreateSession(ctx context.Context, userID, lessonID int64) (*models.WatchSession, error) { + row := s.pool.QueryRow(ctx, ` + INSERT INTO watch_sessions (user_id, lesson_id) VALUES ($1,$2) + ON CONFLICT (user_id, lesson_id) DO UPDATE SET user_id=EXCLUDED.user_id + RETURNING id, user_id, lesson_id, started_at, last_heartbeat_at, last_position`, + userID, lessonID) + var ws models.WatchSession + err := row.Scan(&ws.ID, &ws.UserID, &ws.LessonID, &ws.StartedAt, &ws.LastHeartbeatAt, &ws.LastPosition) + return &ws, err +} + +func (s *Store) InsertHeartbeat(ctx context.Context, hb *models.WatchSession, position, delta float64, seq int) (bool, error) { + tag, err := s.pool.Exec(ctx, ` + INSERT INTO watch_heartbeats (user_id, lesson_id, session_id, position, delta, seq) + VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (user_id, lesson_id, seq) DO NOTHING`, + hb.UserID, hb.LessonID, hb.ID, position, delta, seq) + if err != nil { + return false, err + } + return tag.RowsAffected() > 0, nil +} + +func (s *Store) UpdateSessionPos(ctx context.Context, sessionID int64, position float64) error { + _, err := s.pool.Exec(ctx, ` + UPDATE watch_sessions SET last_position=$1, last_heartbeat_at=now() WHERE id=$2`, position, sessionID) + return err +} + +// MaxHeartbeatSeq returns the highest heartbeat seq for a user/lesson (0 if none). +func (s *Store) MaxHeartbeatSeq(ctx context.Context, userID, lessonID int64) (int, error) { + var seq int + err := s.pool.QueryRow(ctx, ` + SELECT COALESCE(MAX(seq), 0) FROM watch_heartbeats + WHERE user_id=$1 AND lesson_id=$2`, userID, lessonID).Scan(&seq) + return seq, err +} + +func (s *Store) GetProgress(ctx context.Context, userID, lessonID int64) (*models.LessonProgress, error) { + row := s.pool.QueryRow(ctx, ` + SELECT user_id, lesson_id, watched_seconds, watched_pct, last_position, + quiz_unlocked, passed_quiz, quiz_completed_at + FROM lesson_progress WHERE user_id=$1 AND lesson_id=$2`, userID, lessonID) + var p models.LessonProgress + err := row.Scan(&p.UserID, &p.LessonID, &p.WatchedSeconds, &p.WatchedPct, &p.LastPosition, + &p.QuizUnlocked, &p.PassedQuiz, &p.QuizCompletedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return &p, err +} + +func (s *Store) UpsertProgress(ctx context.Context, p *models.LessonProgress) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO lesson_progress (user_id, lesson_id, watched_seconds, watched_pct, last_position) + VALUES ($1,$2,$3,$4,$5) + ON CONFLICT (user_id, lesson_id) DO UPDATE SET + watched_seconds = GREATEST(lesson_progress.watched_seconds, EXCLUDED.watched_seconds), + watched_pct = GREATEST(lesson_progress.watched_pct, EXCLUDED.watched_pct), + last_position = CASE + WHEN EXCLUDED.watched_seconds >= lesson_progress.watched_seconds THEN EXCLUDED.last_position + ELSE lesson_progress.last_position + END`, + p.UserID, p.LessonID, p.WatchedSeconds, p.WatchedPct, p.LastPosition) + return err +} + +func (s *Store) SetQuizUnlocked(ctx context.Context, userID, lessonID int64) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO lesson_progress (user_id, lesson_id, quiz_unlocked) + VALUES ($1,$2,true) + ON CONFLICT (user_id, lesson_id) DO UPDATE SET quiz_unlocked=true`, + userID, lessonID) + return err +} + +func (s *Store) MarkQuizPassed(ctx context.Context, userID, lessonID int64) error { + _, err := s.ClaimFirstQuizPass(ctx, userID, lessonID) + return err +} + +// ClaimFirstQuizPass marks the lesson passed only if it was not already passed. +// Returns true when this call is the first successful pass (XP should be awarded). +func (s *Store) ClaimFirstQuizPass(ctx context.Context, userID, lessonID int64) (bool, error) { + _, _ = s.pool.Exec(ctx, ` + INSERT INTO lesson_progress (user_id, lesson_id) + VALUES ($1,$2) ON CONFLICT (user_id, lesson_id) DO NOTHING`, userID, lessonID) + tag, err := s.pool.Exec(ctx, ` + UPDATE lesson_progress + SET passed_quiz=true, + quiz_unlocked=true, + quiz_completed_at=COALESCE(quiz_completed_at, now()) + WHERE user_id=$1 AND lesson_id=$2 AND NOT passed_quiz`, userID, lessonID) + if err != nil { + return false, err + } + return tag.RowsAffected() > 0, nil +} + +func (s *Store) ProgressMap(ctx context.Context, userID int64) (map[int64]*models.LessonProgress, error) { + rows, err := s.pool.Query(ctx, ` + SELECT user_id, lesson_id, watched_seconds, watched_pct, last_position, + quiz_unlocked, passed_quiz, quiz_completed_at + FROM lesson_progress WHERE user_id=$1`, userID) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int64]*models.LessonProgress{} + for rows.Next() { + var p models.LessonProgress + if err := rows.Scan(&p.UserID, &p.LessonID, &p.WatchedSeconds, &p.WatchedPct, &p.LastPosition, + &p.QuizUnlocked, &p.PassedQuiz, &p.QuizCompletedAt); err != nil { + return nil, err + } + out[p.LessonID] = &p + } + return out, rows.Err() +} + +// ---- Quiz attempts ---- + +func (s *Store) CreateAttempt(ctx context.Context, userID, lessonID int64, answers []int) (*models.QuizAttempt, error) { + b, _ := json.Marshal(answers) + row := s.pool.QueryRow(ctx, ` + INSERT INTO quiz_attempts (user_id, lesson_id, answers) VALUES ($1,$2,$3) + RETURNING id, user_id, lesson_id, answers, correct_count, total, score_pct, status, hearts_lost, passed_at, created_at`, + userID, lessonID, b) + var a models.QuizAttempt + var raw []byte + if err := row.Scan(&a.ID, &a.UserID, &a.LessonID, &raw, &a.CorrectCount, &a.Total, &a.ScorePct, + &a.Status, &a.HeartsLost, &a.PassedAt, &a.CreatedAt); err != nil { + return nil, err + } + _ = json.Unmarshal(raw, &a.Answers) + return &a, nil +} + +func (s *Store) FinalizeAttempt(ctx context.Context, attemptID int64, correct, total int, scorePct float64, status string, heartsLost int) error { + _, err := s.pool.Exec(ctx, ` + UPDATE quiz_attempts SET correct_count=$1, total=$2, score_pct=$3, status=($4)::quiz_status, hearts_lost=$5, + passed_at=CASE WHEN $4='passed' THEN now() ELSE passed_at END + WHERE id=$6`, correct, total, scorePct, status, heartsLost, attemptID) + return err +} + +func (s *Store) LastAttempt(ctx context.Context, userID, lessonID int64) (*models.QuizAttempt, error) { + row := s.pool.QueryRow(ctx, ` + SELECT id, user_id, lesson_id, answers, correct_count, total, score_pct, status, hearts_lost, passed_at, created_at + FROM quiz_attempts WHERE user_id=$1 AND lesson_id=$2 ORDER BY id DESC LIMIT 1`, userID, lessonID) + var a models.QuizAttempt + var raw []byte + if err := row.Scan(&a.ID, &a.UserID, &a.LessonID, &raw, &a.CorrectCount, &a.Total, &a.ScorePct, + &a.Status, &a.HeartsLost, &a.PassedAt, &a.CreatedAt); err != nil { + return nil, err + } + _ = json.Unmarshal(raw, &a.Answers) + return &a, nil +} + +// ---- XP / streaks / leaderboard ---- + +func (s *Store) AddXPEvent(ctx context.Context, tx *Tx, userID int64, amount int, source string, refID int64) error { + _, err := tx.queries.Exec(ctx, `INSERT INTO xp_events (user_id, amount, source, ref_id) VALUES ($1,$2,$3,$4)`, + userID, amount, source, refID) + return err +} + +func (s *Store) UpdateLeaderboardRow(ctx context.Context, week string, userID int64, amount int) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO weekly_leaderboard (week, user_id, xp) VALUES ($1,$2,$3) + ON CONFLICT (week, user_id) DO UPDATE SET xp = weekly_leaderboard.xp + EXCLUDED.xp`, + week, userID, amount) + return err +} + +// ---- Events / RSVPs ---- + +func (s *Store) CreateEvent(ctx context.Context, e *models.Event) (int64, error) { + err := s.pool.QueryRow(ctx, ` + INSERT INTO events (title, description, event_type, external_url, starts_at, ends_at, is_active, created_by) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING id`, + e.Title, e.Description, e.EventType, e.ExternalURL, e.StartsAt, e.EndsAt, e.IsActive, e.CreatedBy).Scan(&e.ID) + return e.ID, err +} + +func (s *Store) UpdateEvent(ctx context.Context, id int64, e *models.Event) error { + _, err := s.pool.Exec(ctx, ` + UPDATE events SET title=$1, description=$2, event_type=$3, external_url=$4, starts_at=$5, ends_at=$6, is_active=$7 + WHERE id=$8`, e.Title, e.Description, e.EventType, e.ExternalURL, e.StartsAt, e.EndsAt, e.IsActive, id) + return err +} + +func (s *Store) DeleteEvent(ctx context.Context, id int64) error { + _, err := s.pool.Exec(ctx, `DELETE FROM events WHERE id=$1`, id) + return err +} + +func (s *Store) GetEvent(ctx context.Context, id int64) (*models.Event, error) { + row := s.pool.QueryRow(ctx, ` + SELECT id, title, description, event_type, external_url, starts_at, ends_at, is_active, created_at + FROM events WHERE id=$1`, id) + var e models.Event + err := row.Scan(&e.ID, &e.Title, &e.Description, &e.EventType, &e.ExternalURL, &e.StartsAt, &e.EndsAt, &e.IsActive, &e.CreatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return &e, err +} + +func (s *Store) ListUpcomingEvents(ctx context.Context, userID int64) ([]models.Event, error) { + rows, err := s.pool.Query(ctx, ` + SELECT e.id, e.title, e.description, e.event_type, e.external_url, e.starts_at, e.ends_at, e.is_active, e.created_at, + CASE WHEN EXISTS (SELECT 1 FROM event_rsvps r WHERE r.event_id=e.id AND r.user_id=$1) THEN true ELSE false END AS rsvped + FROM events e WHERE e.is_active AND e.ends_at > now() ORDER BY e.starts_at ASC`, userID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.Event + for rows.Next() { + var e models.Event + if err := rows.Scan(&e.ID, &e.Title, &e.Description, &e.EventType, &e.ExternalURL, &e.StartsAt, &e.EndsAt, &e.IsActive, &e.CreatedAt, &e.Rsvped); err != nil { + return nil, err + } + out = append(out, e) + } + return out, rows.Err() +} + +func (s *Store) Rsvp(ctx context.Context, eventID, userID int64) error { + _, err := s.pool.Exec(ctx, `INSERT INTO event_rsvps (event_id, user_id) VALUES ($1,$2) ON CONFLICT DO NOTHING`, eventID, userID) + return err +} + +func (s *Store) Unrsvp(ctx context.Context, eventID, userID int64) error { + _, err := s.pool.Exec(ctx, `DELETE FROM event_rsvps WHERE event_id=$1 AND user_id=$2`, eventID, userID) + return err +} + +func (s *Store) EventRsvpUsers(ctx context.Context, eventID int64) ([]models.User, error) { + rows, err := s.pool.Query(ctx, ` + SELECT `+userCols+` FROM users WHERE id IN (SELECT user_id FROM event_rsvps WHERE event_id=$1)`, eventID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.User + for rows.Next() { + u, err := scanUser(rows) + if err != nil { + return nil, err + } + out = append(out, *u) + } + return out, rows.Err() +} + +// ---- Chat ---- + +func (s *Store) AddChatMessage(ctx context.Context, userID, mentorID int64, senderRole models.Role, body string, replyTo *int64, att *models.Attachment) (*models.ChatMessage, error) { + var attType, attURL, attName *string + var attSize *int64 + if att != nil { + attType, attURL, attName = &att.Type, &att.URL, &att.Name + attSize = &att.Size + } + row := s.pool.QueryRow(ctx, ` + INSERT INTO chat_messages (user_id, mentor_id, sender_role, body, reply_to, attachment_type, attachment_url, attachment_name, attachment_size) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) + RETURNING id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at`, + userID, mentorID, senderRole.String(), body, replyTo, attType, attURL, attName, attSize) + return scanMessage(row) +} + +// ImportChatMessage inserts a historical message (backup restore), preserving created_at / pin. +func (s *Store) ImportChatMessage(ctx context.Context, userID, mentorID int64, senderRole models.Role, body string, att *models.Attachment, createdAt time.Time, pinned bool) (*models.ChatMessage, error) { + var attType, attURL, attName *string + var attSize *int64 + if att != nil { + attType, attURL, attName = &att.Type, &att.URL, &att.Name + attSize = &att.Size + } + if createdAt.IsZero() { + createdAt = time.Now().UTC() + } + var pinnedAt *time.Time + if pinned { + t := createdAt + pinnedAt = &t + } + row := s.pool.QueryRow(ctx, ` + INSERT INTO chat_messages (user_id, mentor_id, sender_role, body, attachment_type, attachment_url, attachment_name, attachment_size, created_at, pinned_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) + RETURNING id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at`, + userID, mentorID, senderRole.String(), body, attType, attURL, attName, attSize, createdAt, pinnedAt) + return scanMessage(row) +} + +// ListChatPartnerIDs returns distinct partner user ids for conversations involving me. +func (s *Store) ListChatPartnerIDs(ctx context.Context, meID int64) ([]int64, error) { + rows, err := s.pool.Query(ctx, ` + SELECT DISTINCT partner FROM ( + SELECT user_id AS partner FROM chat_messages WHERE mentor_id=$1 + UNION + SELECT mentor_id FROM chat_messages WHERE user_id=$1 + ) p + WHERE partner <> $1 + ORDER BY partner`, meID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + out = append(out, id) + } + return out, rows.Err() +} + +// DeleteChatMessage removes one of my own messages in this conversation. +func (s *Store) DeleteChatMessage(ctx context.Context, meID, partnerID, msgID int64, meRole models.Role) (bool, error) { + tag, err := s.pool.Exec(ctx, ` + DELETE FROM chat_messages + WHERE id=$1 + AND ((user_id=$2 AND mentor_id=$3) OR (user_id=$3 AND mentor_id=$2)) + AND sender_role=$4 + AND (user_id=$2 OR mentor_id=$2)`, msgID, meID, partnerID, meRole.String()) + if err != nil { + return false, err + } + return tag.RowsAffected() > 0, nil +} + +const maxPinnedMessages = 3 + +// TogglePinChatMessage pins or unpins a message. If already at the pin limit, +// the oldest pin is replaced so a new message (including emoji-only) can be pinned. +func (s *Store) TogglePinChatMessage(ctx context.Context, meID, partnerID, msgID int64) (*models.ChatMessage, error) { + ok, err := s.MessageInConversation(ctx, msgID, meID, partnerID) + if err != nil { + return nil, err + } + if !ok { + return nil, ErrNotFound + } + row := s.pool.QueryRow(ctx, ` + SELECT pinned_at FROM chat_messages WHERE id=$1`, msgID) + var pinnedAt *time.Time + if err := row.Scan(&pinnedAt); err != nil { + return nil, err + } + if pinnedAt != nil { + out := s.pool.QueryRow(ctx, ` + UPDATE chat_messages SET pinned_at=NULL WHERE id=$1 + RETURNING id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at`, msgID) + return scanMessage(out) + } + var n int + if err := s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM chat_messages + WHERE pinned_at IS NOT NULL + AND ((user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1))`, + meID, partnerID).Scan(&n); err != nil { + return nil, err + } + if n >= maxPinnedMessages { + // Evict oldest pin in this conversation, then pin the new message. + if _, err := s.pool.Exec(ctx, ` + UPDATE chat_messages SET pinned_at=NULL + WHERE id = ( + SELECT id FROM chat_messages + WHERE pinned_at IS NOT NULL + AND ((user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1)) + ORDER BY pinned_at ASC + LIMIT 1 + )`, meID, partnerID); err != nil { + return nil, err + } + } + out := s.pool.QueryRow(ctx, ` + UPDATE chat_messages SET pinned_at=now() WHERE id=$1 + RETURNING id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at`, msgID) + return scanMessage(out) +} + +// ListPinnedChatMessages returns pinned messages for a conversation (newest pin first). +func (s *Store) ListPinnedChatMessages(ctx context.Context, userID, mentorID int64) ([]models.ChatMessage, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at + FROM chat_messages + WHERE pinned_at IS NOT NULL + AND ((user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1)) + ORDER BY pinned_at DESC + LIMIT $3`, userID, mentorID, maxPinnedMessages) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.ChatMessage + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, err + } + out = append(out, *m) + } + return out, rows.Err() +} + +// ListChatMessages returns the most recent `limit` messages strictly older than +// `before` (id cursor, 0 = no cutoff), ordered oldest-first. +func (s *Store) ListChatMessages(ctx context.Context, userID, mentorID, before, limit int64) ([]models.ChatMessage, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at + FROM chat_messages + WHERE ((user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1)) + AND ($3 = 0 OR id < $3) + ORDER BY id DESC + LIMIT $4`, userID, mentorID, before, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.ChatMessage + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, err + } + out = append(out, *m) + } + if err := rows.Err(); err != nil { + return nil, err + } + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + return out, nil +} + +// ChatMessageInPair reports whether msgID belongs to the conversation between the two users. +func (s *Store) ChatMessageInPair(ctx context.Context, userID, partnerID, msgID int64) (bool, error) { + var ok bool + err := s.pool.QueryRow(ctx, ` + SELECT EXISTS( + SELECT 1 FROM chat_messages + WHERE id=$3 AND ((user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1)) + )`, userID, partnerID, msgID).Scan(&ok) + return ok, err +} + +// ListChatMessagesAround returns a window of messages centered on aroundID (oldest-first). +func (s *Store) ListChatMessagesAround(ctx context.Context, userID, partnerID, aroundID, limit int64) (msgs []models.ChatMessage, hasMoreBefore, hasMoreAfter bool, err error) { + if limit < 4 { + limit = 60 + } + beforeLimit := limit/2 + 1 // include target + afterLimit := limit / 2 + + older, err := s.pool.Query(ctx, ` + SELECT id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at + FROM chat_messages + WHERE ((user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1)) + AND id <= $3 + ORDER BY id DESC + LIMIT $4`, userID, partnerID, aroundID, beforeLimit+1) + if err != nil { + return nil, false, false, err + } + var before []models.ChatMessage + for older.Next() { + m, scanErr := scanMessage(older) + if scanErr != nil { + older.Close() + return nil, false, false, scanErr + } + before = append(before, *m) + } + older.Close() + if err := older.Err(); err != nil { + return nil, false, false, err + } + if int64(len(before)) > beforeLimit { + hasMoreBefore = true + before = before[:beforeLimit] + } + for i, j := 0, len(before)-1; i < j; i, j = i+1, j-1 { + before[i], before[j] = before[j], before[i] + } + + newer, err := s.pool.Query(ctx, ` + SELECT id, user_id, mentor_id, sender_role, body, read_at, created_at, reply_to, + attachment_type, attachment_url, attachment_name, attachment_size, pinned_at + FROM chat_messages + WHERE ((user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1)) + AND id > $3 + ORDER BY id ASC + LIMIT $4`, userID, partnerID, aroundID, afterLimit+1) + if err != nil { + return nil, false, false, err + } + var after []models.ChatMessage + for newer.Next() { + m, scanErr := scanMessage(newer) + if scanErr != nil { + newer.Close() + return nil, false, false, scanErr + } + after = append(after, *m) + } + newer.Close() + if err := newer.Err(); err != nil { + return nil, false, false, err + } + if int64(len(after)) > afterLimit { + hasMoreAfter = true + after = after[:afterLimit] + } + + out := make([]models.ChatMessage, 0, len(before)+len(after)) + out = append(out, before...) + out = append(out, after...) + return out, hasMoreBefore, hasMoreAfter, nil +} + +type messageScanner interface { + Scan(dest ...any) error +} + +func scanMessage(row messageScanner) (*models.ChatMessage, error) { + var m models.ChatMessage + var attType, attURL, attName *string + var attSize *int64 + if err := row.Scan( + &m.ID, &m.UserID, &m.MentorID, &m.SenderRole, &m.Body, &m.ReadAt, &m.CreatedAt, &m.ReplyTo, + &attType, &attURL, &attName, &attSize, &m.PinnedAt, + ); err != nil { + return nil, err + } + m.Pinned = m.PinnedAt != nil + if attURL != nil && *attURL != "" { + m.Attachment = &models.Attachment{ + Type: derefStr(attType), URL: *attURL, Name: derefStr(attName), Size: derefInt(attSize), + } + } + return &m, nil +} + +func derefStr(p *string) string { + if p == nil { + return "" + } + return *p +} + +func derefInt(p *int64) int64 { + if p == nil { + return 0 + } + return *p +} + +func (s *Store) MarkChatRead(ctx context.Context, userID, mentorID int64) error { + _, err := s.pool.Exec(ctx, ` + UPDATE chat_messages SET read_at=now() WHERE (user_id=$1 AND mentor_id=$2) OR (user_id=$2 AND mentor_id=$1)`, + userID, mentorID) + return err +} + +// MessageInConversation reports whether a message belongs to a given pair. +func (s *Store) MessageInConversation(ctx context.Context, msgID, meID, partnerID int64) (bool, error) { + var one int + err := s.pool.QueryRow(ctx, ` + SELECT 1 FROM chat_messages + WHERE id=$1 AND ((user_id=$2 AND mentor_id=$3) OR (user_id=$3 AND mentor_id=$2))`, + msgID, meID, partnerID).Scan(&one) + if err == pgx.ErrNoRows { + return false, nil + } + return err == nil, err +} + +// ReactToChatMessage toggles a reaction on a message: same emoji removes it, +// a different emoji replaces it, a new one is inserted. Returns the resulting +// emoji ("" when removed) and the full updated set. +func (s *Store) ReactToChatMessage(ctx context.Context, msgID, userID int64, emoji string) (string, []models.ChatReaction, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return "", nil, err + } + defer func() { _ = tx.Rollback(ctx) }() + + var existing string + err = tx.QueryRow(ctx, `SELECT emoji FROM chat_reactions WHERE message_id=$1 AND user_id=$2`, msgID, userID).Scan(&existing) + switch { + case err == nil && existing == emoji: + if _, err := tx.Exec(ctx, `DELETE FROM chat_reactions WHERE message_id=$1 AND user_id=$2`, msgID, userID); err != nil { + return "", nil, err + } + emoji = "" + case err == nil: + if _, err := tx.Exec(ctx, `UPDATE chat_reactions SET emoji=$3, created_at=now() WHERE message_id=$1 AND user_id=$2`, msgID, userID, emoji); err != nil { + return "", nil, err + } + case errors.Is(err, pgx.ErrNoRows): + if _, err := tx.Exec(ctx, `INSERT INTO chat_reactions (message_id, user_id, emoji) VALUES ($1,$2,$3)`, msgID, userID, emoji); err != nil { + return "", nil, err + } + default: + return "", nil, err + } + if err := tx.Commit(ctx); err != nil { + return "", nil, err + } + reactions, err := s.ReactionsFor(ctx, msgID, userID) + if err != nil { + return "", nil, err + } + return emoji, reactions, nil +} + +// ReactionsFor returns the aggregated reactions for a single message. +func (s *Store) ReactionsFor(ctx context.Context, msgID, meID int64) ([]models.ChatReaction, error) { + rows, err := s.pool.Query(ctx, ` + SELECT emoji, count(*)::int, bool_or(user_id=$2) AS mine + FROM chat_reactions WHERE message_id=$1 + GROUP BY emoji ORDER BY count(*) DESC, emoji`, msgID, meID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []models.ChatReaction + for rows.Next() { + var r models.ChatReaction + if err := rows.Scan(&r.Emoji, &r.Count, &r.Mine); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// FillReactions attaches aggregated reactions to the given messages (one query). +func (s *Store) FillReactions(ctx context.Context, meID int64, msgs []models.ChatMessage) error { + if len(msgs) == 0 { + return nil + } + ids := make([]int64, 0, len(msgs)) + for _, m := range msgs { + if m.ID > 0 { + ids = append(ids, m.ID) + } + } + if len(ids) == 0 { + return nil + } + rows, err := s.pool.Query(ctx, ` + SELECT message_id, emoji, count(*)::int, bool_or(user_id=$1) AS mine + FROM chat_reactions WHERE message_id = ANY($2) + GROUP BY message_id, emoji + ORDER BY count(*) DESC, emoji`, meID, ids) + if err != nil { + return err + } + defer rows.Close() + byMsg := map[int64][]models.ChatReaction{} + for rows.Next() { + var msgID int64 + var r models.ChatReaction + if err := rows.Scan(&msgID, &r.Emoji, &r.Count, &r.Mine); err != nil { + return err + } + byMsg[msgID] = append(byMsg[msgID], r) + } + if err := rows.Err(); err != nil { + return err + } + for i := range msgs { + if msgs[i].ID > 0 { + msgs[i].Reactions = byMsg[msgs[i].ID] + } + } + return nil +} + +// ---- Notifications ---- + +func (s *Store) Notify(ctx context.Context, userID int64, category, typeStr, title, body, route string, data any) error { + d, _ := json.Marshal(data) + _, err := s.pool.Exec(ctx, ` + INSERT INTO notifications (user_id, category, type, title, body, route, data) + VALUES ($1,$2,$3,$4,$5,$6,$7)`, + userID, category, typeStr, title, body, route, d) + return err +} + +func (s *Store) CreateNotification(ctx context.Context, n *models.Notification) error { + d := n.Data + if d == nil { + d = json.RawMessage(`{}`) + } + err := s.pool.QueryRow(ctx, ` + INSERT INTO notifications (user_id, category, type, title, body, route, data) + VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id, created_at`, + n.UserID, n.Category, n.Type, n.Title, n.Body, n.Route, d).Scan(&n.ID, &n.CreatedAt) + return err +} + +func (s *Store) GetPrefs(ctx context.Context, userID int64) (*models.NotificationPrefs, error) { + if _, err := s.pool.Exec(ctx, ` + INSERT INTO notification_prefs (user_id) VALUES ($1) + ON CONFLICT (user_id) DO NOTHING`, userID); err != nil { + return nil, err + } + var p models.NotificationPrefs + if err := s.pool.QueryRow(ctx, ` + SELECT user_id, progress, gamification, mentor, event FROM notification_prefs WHERE user_id=$1`, userID). + Scan(&p.UserID, &p.Progress, &p.Gamification, &p.Mentor, &p.Event); err != nil { + return nil, err + } + return &p, nil +} + +func (s *Store) UpdatePrefs(ctx context.Context, p *models.NotificationPrefs) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO notification_prefs (user_id, progress, gamification, mentor, event) + VALUES ($1,$2,$3,$4,$5) + ON CONFLICT (user_id) DO UPDATE SET progress=$2, gamification=$3, mentor=$4, event=$5`, + p.UserID, p.Progress, p.Gamification, p.Mentor, p.Event) + return err +} + +func (s *Store) PrefEnabled(ctx context.Context, userID int64, category string) (bool, error) { + var v bool + err := s.pool.QueryRow(ctx, ` + SELECT CASE $2 + WHEN 'progress' THEN progress + WHEN 'gamification' THEN gamification + WHEN 'mentor' THEN mentor + WHEN 'event' THEN event + END FROM notification_prefs WHERE user_id=$1`, userID, category).Scan(&v) + if err != nil { + return true, nil + } + return v, nil +} + +func (s *Store) ListNotifications(ctx context.Context, userID int64, limit, offset int) ([]models.Notification, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, user_id, category, type, title, body, route, data, read_at, created_at + FROM notifications WHERE user_id=$1 ORDER BY id DESC LIMIT $2 OFFSET $3`, userID, limit, offset) + if err != nil { + return nil, err + } + defer rows.Close() + return scanNotifications(rows) +} + +func scanNotifications(rows pgx.Rows) ([]models.Notification, error) { + var out []models.Notification + for rows.Next() { + var n models.Notification + if err := rows.Scan(&n.ID, &n.UserID, &n.Category, &n.Type, &n.Title, &n.Body, &n.Route, &n.Data, &n.ReadAt, &n.CreatedAt); err != nil { + return nil, err + } + out = append(out, n) + } + return out, rows.Err() +} + +func (s *Store) UnreadCount(ctx context.Context, userID int64) (int, error) { + var n int + err := s.pool.QueryRow(ctx, `SELECT count(*) FROM notifications WHERE user_id=$1 AND read_at IS NULL`, userID).Scan(&n) + return n, err +} + +func (s *Store) MarkRead(ctx context.Context, userID, notifID int64) error { + _, err := s.pool.Exec(ctx, `UPDATE notifications SET read_at=now() WHERE id=$1 AND user_id=$2`, notifID, userID) + return err +} + +func (s *Store) MarkAllRead(ctx context.Context, userID int64) error { + _, err := s.pool.Exec(ctx, `UPDATE notifications SET read_at=now() WHERE user_id=$1 AND read_at IS NULL`, userID) + return err +} + +// ---- Reminders ---- + +func (s *Store) ReminderSent(ctx context.Context, userID int64, kind, ref string) (bool, error) { + var one int + err := s.pool.QueryRow(ctx, `SELECT 1 FROM reminders_sent WHERE user_id=$1 AND kind=$2 AND ref=$3`, userID, kind, ref).Scan(&one) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } + return true, err +} + +func (s *Store) MarkReminderSent(ctx context.Context, userID int64, kind, ref string) error { + _, err := s.pool.Exec(ctx, `INSERT INTO reminders_sent (user_id, kind, ref) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING`, userID, kind, ref) + return err +} + +// ---- Video assets ---- + +func (s *Store) CreateVideoAsset(ctx context.Context, key, mime string, size int64) error { + _, err := s.pool.Exec(ctx, `INSERT INTO video_assets (key, mime, size_bytes) VALUES ($1,$2,$3) ON CONFLICT (key) DO NOTHING`, key, mime, size) + return err +} + +type VideoAssetRow struct { + Key string + Mime string + SizeBytes int64 + CreatedAt time.Time +} + +func (s *Store) ListVideoAssets(ctx context.Context) ([]VideoAssetRow, error) { + rows, err := s.pool.Query(ctx, ` + SELECT key, mime, size_bytes, created_at + FROM video_assets + ORDER BY created_at DESC + LIMIT 500`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []VideoAssetRow + for rows.Next() { + var v VideoAssetRow + if err := rows.Scan(&v.Key, &v.Mime, &v.SizeBytes, &v.CreatedAt); err != nil { + return nil, err + } + out = append(out, v) + } + if out == nil { + out = []VideoAssetRow{} + } + return out, rows.Err() +} + +func (s *Store) DeleteVideoAsset(ctx context.Context, key string) error { + _, err := s.pool.Exec(ctx, `DELETE FROM video_assets WHERE key=$1`, key) + return err +} + +// ---- Refresh tokens ---- + +func (s *Store) CreateRefreshToken(ctx context.Context, userID int64, hash string, expiresAt time.Time) error { + _, err := s.pool.Exec(ctx, `INSERT INTO refresh_tokens (user_id, token_hash, expires_at) VALUES ($1,$2,$3)`, userID, hash, expiresAt) + return err +} + +func (s *Store) RevokeRefreshToken(ctx context.Context, hash string) error { + _, err := s.pool.Exec(ctx, `UPDATE refresh_tokens SET revoked_at=now() WHERE token_hash=$1`, hash) + return err +} + +func (s *Store) RevokeAllUserRefreshTokens(ctx context.Context, userID int64) error { + _, err := s.pool.Exec(ctx, ` + UPDATE refresh_tokens SET revoked_at=now() + WHERE user_id=$1 AND revoked_at IS NULL`, userID) + return err +} + +// ConsumeRefreshToken atomically revokes a valid refresh token and returns its +// user id plus the original lifetime (expires_at − created_at) so callers can +// preserve remember-me vs short session on rotation. +func (s *Store) ConsumeRefreshToken(ctx context.Context, hash string) (userID int64, lifetime time.Duration, ok bool, err error) { + var expiresAt, createdAt time.Time + err = s.pool.QueryRow(ctx, ` + UPDATE refresh_tokens + SET revoked_at=now() + WHERE token_hash=$1 AND revoked_at IS NULL AND expires_at > now() + RETURNING user_id, expires_at, created_at`, hash).Scan(&userID, &expiresAt, &createdAt) + if errors.Is(err, pgx.ErrNoRows) { + return 0, 0, false, nil + } + if err != nil { + return 0, 0, false, err + } + lifetime = expiresAt.Sub(createdAt) + if lifetime < 0 { + lifetime = 0 + } + return userID, lifetime, true, nil +} + +func (s *Store) RefreshTokenValid(ctx context.Context, hash string) (int64, bool, error) { + var userID int64 + var expiresAt, revokedAt *time.Time + err := s.pool.QueryRow(ctx, ` + SELECT user_id, expires_at, revoked_at FROM refresh_tokens WHERE token_hash=$1`, hash). + Scan(&userID, &expiresAt, &revokedAt) + if errors.Is(err, pgx.ErrNoRows) { + return 0, false, nil + } + if err != nil { + return 0, false, err + } + ok := (revokedAt == nil) && (expiresAt != nil && expiresAt.After(time.Now())) + return userID, ok, nil +} + +func (s *Store) PurgeOldHeartbeats(ctx context.Context, olderThan time.Duration) (int64, error) { + cutoff := time.Now().Add(-olderThan) + tag, err := s.pool.Exec(ctx, `DELETE FROM watch_heartbeats WHERE created_at < $1`, cutoff) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + +// ---- Announcements / stats ---- + +func (s *Store) AllActiveUserIDs(ctx context.Context) ([]int64, error) { + rows, err := s.pool.Query(ctx, `SELECT id FROM users WHERE is_active`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + out = append(out, id) + } + return out, rows.Err() +} + +func (s *Store) Stats(ctx context.Context) (map[string]any, error) { + var users, lessons, questions, events, completions int64 + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM users`).Scan(&users); err != nil { + return nil, err + } + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM lessons`).Scan(&lessons); err != nil { + return nil, err + } + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM mcq_questions`).Scan(&questions); err != nil { + return nil, err + } + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM events`).Scan(&events); err != nil { + return nil, err + } + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM lesson_progress WHERE passed_quiz`).Scan(&completions); err != nil { + return nil, err + } + return map[string]any{ + "users": users, "lessons": lessons, "questions": questions, + "events": events, "completions": completions, + }, nil +} diff --git a/backend/internal/store/whitelist.go b/backend/internal/store/whitelist.go new file mode 100644 index 0000000..15cbf11 --- /dev/null +++ b/backend/internal/store/whitelist.go @@ -0,0 +1,238 @@ +package store + +import ( + "context" + "errors" + "strconv" + "strings" + + "github.com/jackc/pgx/v5" + + "pargar/backend/internal/models" +) + +var ( + ErrWhitelistNotAllowed = errors.New("phone not on whitelist") + ErrWhitelistConsumed = errors.New("whitelist phone already consumed") + ErrWhitelistNotFound = errors.New("whitelist entry not found") + ErrWhitelistNotAvailable = errors.New("whitelist entry not available") +) + +const whitelistCols = `id, phone, status, consumed_at, consumed_user_id, created_by, created_at` + +func scanWhitelist(row rowScanner) (*models.RegistrationPhoneWhitelist, error) { + var e models.RegistrationPhoneWhitelist + var status string + err := row.Scan( + &e.ID, &e.Phone, &status, &e.ConsumedAt, &e.ConsumedUserID, &e.CreatedBy, &e.CreatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrWhitelistNotFound + } + if err != nil { + return nil, err + } + e.Status = models.WhitelistStatus(status) + return &e, nil +} + +// WhitelistImportResult is the outcome of a whitelist import batch. +type WhitelistImportResult struct { + Inserted int + Skipped int + Rejected int + RejectedPhones []string + RemovedBlacklist int + BlacklistedConflicts []string +} + +// ImportWhitelistPhones inserts available phones; skips duplicates. +// Phones on the blacklist are reported in BlacklistedConflicts unless resolveConflicts +// is true, in which case they are removed from the blacklist then added to the whitelist. +// Phones already belonging to a user are rejected (cannot be on both lists / re-registered). +func (s *Store) ImportWhitelistPhones(ctx context.Context, phones []string, createdBy int64, resolveConflicts bool) (WhitelistImportResult, error) { + var result WhitelistImportResult + if len(phones) == 0 { + return result, nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return result, err + } + defer tx.Rollback(ctx) + + for _, phone := range phones { + phone = strings.TrimSpace(phone) + if phone == "" { + continue + } + + var userExists bool + if err := tx.QueryRow(ctx, ` + SELECT EXISTS(SELECT 1 FROM users WHERE phone=$1)`, phone).Scan(&userExists); err != nil { + return WhitelistImportResult{}, err + } + if userExists { + result.Rejected++ + result.RejectedPhones = append(result.RejectedPhones, phone) + continue + } + + blocked, err := isPhoneBlacklistedTx(ctx, tx, phone) + if err != nil { + return WhitelistImportResult{}, err + } + if blocked { + if !resolveConflicts { + result.BlacklistedConflicts = append(result.BlacklistedConflicts, phone) + continue + } + tag, err := tx.Exec(ctx, ` + DELETE FROM registration_phone_blacklist WHERE phone = $1`, phone) + if err != nil { + return WhitelistImportResult{}, err + } + if tag.RowsAffected() > 0 { + result.RemovedBlacklist++ + } + } + + tag, err := tx.Exec(ctx, ` + INSERT INTO registration_phone_whitelist (phone, created_by) + VALUES ($1, $2) + ON CONFLICT (phone) DO NOTHING`, phone, createdBy) + if err != nil { + return WhitelistImportResult{}, err + } + if tag.RowsAffected() == 0 { + result.Skipped++ + } else { + result.Inserted++ + } + } + if err := assertPhoneListsExclusiveTx(ctx, tx); err != nil { + return WhitelistImportResult{}, err + } + if err := tx.Commit(ctx); err != nil { + return WhitelistImportResult{}, err + } + return result, nil +} + +func (s *Store) ListWhitelistPhones(ctx context.Context, status, query string, limit, offset int) ([]models.RegistrationPhoneWhitelist, int64, error) { + if limit <= 0 { + limit = 10 + } + if limit > 100 { + limit = 100 + } + if offset < 0 { + offset = 0 + } + status = strings.TrimSpace(status) + query = strings.TrimSpace(query) + filtered := status == string(models.WhitelistAvailable) || status == string(models.WhitelistConsumed) + + where := `($1 = '' OR phone ILIKE '%'||$1||'%')` + args := []any{query} + if filtered { + args = append(args, status) + where += ` AND status = $2` + } + + var total int64 + if err := s.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM registration_phone_whitelist WHERE `+where, args...).Scan(&total); err != nil { + return nil, 0, err + } + + limitIdx := len(args) + 1 + offsetIdx := len(args) + 2 + args = append(args, limit, offset) + order := `ORDER BY created_at DESC` + if !filtered { + order = `ORDER BY CASE status WHEN 'available' THEN 0 ELSE 1 END, created_at DESC` + } + rows, err := s.pool.Query(ctx, ` + SELECT `+whitelistCols+` + FROM registration_phone_whitelist + WHERE `+where+` + `+order+` + LIMIT $`+strconv.Itoa(limitIdx)+` OFFSET $`+strconv.Itoa(offsetIdx), args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + var out []models.RegistrationPhoneWhitelist + for rows.Next() { + e, err := scanWhitelist(rows) + if err != nil { + return nil, 0, err + } + out = append(out, *e) + } + return out, total, rows.Err() +} + +func (s *Store) GetWhitelistByPhone(ctx context.Context, phone string) (*models.RegistrationPhoneWhitelist, error) { + row := s.pool.QueryRow(ctx, ` + SELECT `+whitelistCols+` + FROM registration_phone_whitelist WHERE phone=$1`, phone) + return scanWhitelist(row) +} + +// DeleteAvailableWhitelistPhone removes an unused whitelist entry. +func (s *Store) DeleteAvailableWhitelistPhone(ctx context.Context, id int64) error { + tag, err := s.pool.Exec(ctx, ` + DELETE FROM registration_phone_whitelist + WHERE id = $1 AND status = 'available'`, id) + if err != nil { + return err + } + if tag.RowsAffected() > 0 { + return nil + } + var status string + err = s.pool.QueryRow(ctx, ` + SELECT status FROM registration_phone_whitelist WHERE id=$1`, id).Scan(&status) + if errors.Is(err, pgx.ErrNoRows) { + return ErrWhitelistNotFound + } + if err != nil { + return err + } + return ErrWhitelistNotAvailable +} + +// consumeWhitelistPhoneTx marks an available phone as consumed for userID. +// require: if true, missing/consumed returns an error; if false, missing/consumed is a no-op. +func consumeWhitelistPhoneTx(ctx context.Context, tx pgx.Tx, phone string, userID int64, require bool) error { + tag, err := tx.Exec(ctx, ` + UPDATE registration_phone_whitelist + SET status = 'consumed', + consumed_at = now(), + consumed_user_id = $2 + WHERE phone = $1 AND status = 'available'`, phone, userID) + if err != nil { + return err + } + if tag.RowsAffected() > 0 { + return nil + } + if !require { + return nil + } + var status string + err = tx.QueryRow(ctx, ` + SELECT status FROM registration_phone_whitelist WHERE phone=$1`, phone).Scan(&status) + if errors.Is(err, pgx.ErrNoRows) { + return ErrWhitelistNotAllowed + } + if err != nil { + return err + } + if status == string(models.WhitelistConsumed) { + return ErrWhitelistConsumed + } + return ErrWhitelistNotAllowed +}