feat: sync durable moderation and appeals

This commit is contained in:
iamxvbaba 2026-07-24 11:56:59 +08:00
parent e1a95c7318
commit 9f467f4be7
140 changed files with 13730 additions and 316 deletions

View file

@ -26,6 +26,13 @@ type AccountSettingsStore interface {
SaveAccountSettings(ctx context.Context, userID int64, settings domain.AccountSettings) error
}
// AccountSettingsBatchStore is the cold-load boundary for the per-user account
// settings read model. Production stores implement it so a 100-user
// users.getRequirementsToContact request never becomes 100 SQL queries.
type AccountSettingsBatchStore interface {
GetAccountSettingsBatch(ctx context.Context, userIDs []int64) (map[int64]domain.AccountSettings, error)
}
// NotifySettingsStore persists per-scope notification settings (specific peer /
// forum topic / the three category defaults: users, chats, broadcasts).
type NotifySettingsStore interface {

View file

@ -0,0 +1,13 @@
package store
import (
"context"
"time"
"telesrv/internal/domain"
)
type AuthDeliveryReportStore interface {
CreateAuthDeliveryReport(ctx context.Context, report domain.AuthDeliveryReport) (domain.AuthDeliveryReport, bool, error)
DeleteExpiredAuthDeliveryReports(ctx context.Context, olderThan time.Time, limit int) (int, error)
}

View file

@ -0,0 +1,13 @@
package store
import (
"context"
"time"
"telesrv/internal/domain"
)
type ClientTelemetryStore interface {
CreateClientTelemetry(ctx context.Context, event domain.ClientTelemetryEvent) (domain.ClientTelemetryEvent, bool, error)
DeleteExpiredClientTelemetry(ctx context.Context, olderThan time.Time, limit int) (int, error)
}

View file

@ -0,0 +1,83 @@
package memory
import (
"context"
"sync"
"time"
"telesrv/internal/domain"
)
type AuthDeliveryReportStore struct {
mu sync.Mutex
nextID int64
byFingerprint map[[32]byte]domain.AuthDeliveryReport
}
func NewAuthDeliveryReportStore() *AuthDeliveryReportStore {
return &AuthDeliveryReportStore{
nextID: 1, byFingerprint: make(map[[32]byte]domain.AuthDeliveryReport),
}
}
func (s *AuthDeliveryReportStore) CreateAuthDeliveryReport(_ context.Context, report domain.AuthDeliveryReport) (domain.AuthDeliveryReport, bool, error) {
if err := report.Validate(); err != nil {
return domain.AuthDeliveryReport{}, false, err
}
s.mu.Lock()
defer s.mu.Unlock()
if existing, ok := s.byFingerprint[report.Fingerprint]; ok {
return existing, false, nil
}
var hourly, phoneDaily int
hourAgo := report.CreatedAt.Add(-time.Hour)
dayAgo := report.CreatedAt.Add(-24 * time.Hour)
for _, existing := range s.byFingerprint {
if existing.CreatedAt.After(report.CreatedAt) {
continue
}
if existing.AuthKeyID == report.AuthKeyID && !existing.CreatedAt.Before(hourAgo) {
hourly++
}
if existing.PhoneHash == report.PhoneHash && !existing.CreatedAt.Before(dayAgo) {
phoneDaily++
}
}
if hourly >= domain.MaxAuthDeliveryReportsPerHour ||
phoneDaily >= domain.MaxAuthDeliveryReportsPerPhoneDay {
return domain.AuthDeliveryReport{}, false, domain.ErrAuthDeliveryRateLimited
}
report.ID = s.nextID
s.nextID++
s.byFingerprint[report.Fingerprint] = report
return report, true, nil
}
func (s *AuthDeliveryReportStore) Reports() []domain.AuthDeliveryReport {
s.mu.Lock()
defer s.mu.Unlock()
out := make([]domain.AuthDeliveryReport, 0, len(s.byFingerprint))
for _, report := range s.byFingerprint {
out = append(out, report)
}
return out
}
func (s *AuthDeliveryReportStore) DeleteExpiredAuthDeliveryReports(_ context.Context, olderThan time.Time, limit int) (int, error) {
if olderThan.IsZero() || limit <= 0 || limit > 10000 {
return 0, domain.ErrAuthDeliveryReportInvalid
}
s.mu.Lock()
defer s.mu.Unlock()
deleted := 0
for fingerprint, report := range s.byFingerprint {
if deleted >= limit {
break
}
if report.CreatedAt.Before(olderThan) {
delete(s.byFingerprint, fingerprint)
deleted++
}
}
return deleted, nil
}

View file

@ -479,6 +479,39 @@ func (s *ChannelStore) ListChannelMessageReactions(_ context.Context, req domain
}, nil
}
func (s *ChannelStore) FindChannelMessageReaction(_ context.Context, req domain.ChannelMessageReactionLookupRequest) (domain.ChannelMessageReactionLookup, bool, error) {
if req.ViewerUserID == 0 || req.ChannelID == 0 || req.MessageID <= 0 ||
req.MessageID > domain.MaxMessageBoxID || req.ReactorUserID == 0 {
return domain.ChannelMessageReactionLookup{}, false, domain.ErrChannelInvalid
}
s.mu.RLock()
defer s.mu.RUnlock()
channel, member, err := s.channelAndMemberLocked(req.ViewerUserID, req.ChannelID)
if err != nil {
return domain.ChannelMessageReactionLookup{}, false, err
}
message, ok := s.findMessageLocked(req.ChannelID, req.MessageID)
if !ok || message.Deleted || message.ID <= member.AvailableMinID {
return domain.ChannelMessageReactionLookup{}, false, domain.ErrMessageIDInvalid
}
rows := cloneChannelPeerReactions(s.reactions[req.ChannelID][req.MessageID][req.ReactorUserID])
if len(rows) == 0 {
return domain.ChannelMessageReactionLookup{
Channel: cloneChannel(channel), Message: cloneChannelMessage(message),
}, false, nil
}
sort.Slice(rows, func(i, j int) bool {
if rows[i].ChosenOrder != rows[j].ChosenOrder {
return rows[i].ChosenOrder < rows[j].ChosenOrder
}
return messageReactionKey(rows[i].Reaction) < messageReactionKey(rows[j].Reaction)
})
return domain.ChannelMessageReactionLookup{
Channel: cloneChannel(channel), Message: cloneChannelMessage(message),
Reactions: rows,
}, true, nil
}
func (s *ChannelStore) RecordMessageReactionUse(_ context.Context, userID int64, reactions []domain.MessageReaction, addToRecent bool, date int) error {
if userID == 0 || len(reactions) == 0 {
return nil

View file

@ -0,0 +1,99 @@
package memory
import (
"context"
"sort"
"sync"
"time"
"telesrv/internal/domain"
)
type ClientTelemetryStore struct {
mu sync.Mutex
nextID int64
byID map[int64]domain.ClientTelemetryEvent
byFingerprint map[[32]byte]int64
}
func NewClientTelemetryStore() *ClientTelemetryStore {
return &ClientTelemetryStore{
nextID: 1, byID: make(map[int64]domain.ClientTelemetryEvent),
byFingerprint: make(map[[32]byte]int64),
}
}
func (s *ClientTelemetryStore) CreateClientTelemetry(_ context.Context, event domain.ClientTelemetryEvent) (domain.ClientTelemetryEvent, bool, error) {
if err := event.Validate(); err != nil || event.ID != 0 {
return domain.ClientTelemetryEvent{}, false, domain.ErrClientTelemetryInvalid
}
s.mu.Lock()
defer s.mu.Unlock()
if id, ok := s.byFingerprint[event.Fingerprint]; ok {
return cloneClientTelemetry(s.byID[id]), false, nil
}
var hourly, daily int
for _, existing := range s.byID {
if existing.UserID != event.UserID ||
existing.CreatedAt.After(event.CreatedAt) {
continue
}
if !existing.CreatedAt.Before(event.CreatedAt.Add(-24 * time.Hour)) {
daily++
}
if !existing.CreatedAt.Before(event.CreatedAt.Add(-time.Hour)) {
hourly++
}
}
if hourly >= domain.MaxClientTelemetryEventsPerHour ||
daily >= domain.MaxClientTelemetryEventsPerDay {
return domain.ClientTelemetryEvent{}, false, domain.ErrClientTelemetryRateLimited
}
event.ID = s.nextID
s.nextID++
event = cloneClientTelemetry(event)
s.byID[event.ID] = event
s.byFingerprint[event.Fingerprint] = event.ID
return cloneClientTelemetry(event), true, nil
}
func (s *ClientTelemetryStore) DeleteExpiredClientTelemetry(_ context.Context, olderThan time.Time, limit int) (int, error) {
if olderThan.IsZero() || limit <= 0 || limit > 10000 {
return 0, domain.ErrClientTelemetryInvalid
}
s.mu.Lock()
defer s.mu.Unlock()
ids := make([]int64, 0)
for id, event := range s.byID {
if event.CreatedAt.Before(olderThan) {
ids = append(ids, id)
}
}
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
if len(ids) > limit {
ids = ids[:limit]
}
for _, id := range ids {
event := s.byID[id]
delete(s.byFingerprint, event.Fingerprint)
delete(s.byID, id)
}
return len(ids), nil
}
func (s *ClientTelemetryStore) Events() []domain.ClientTelemetryEvent {
s.mu.Lock()
defer s.mu.Unlock()
out := make([]domain.ClientTelemetryEvent, 0, len(s.byID))
for _, event := range s.byID {
out = append(out, cloneClientTelemetry(event))
}
sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
return out
}
func cloneClientTelemetry(event domain.ClientTelemetryEvent) domain.ClientTelemetryEvent {
event.SubjectIDs = append([]int64(nil), event.SubjectIDs...)
event.Payload = append([]byte(nil), event.Payload...)
return event
}

View file

@ -0,0 +1,62 @@
package memory
import (
"context"
"errors"
"testing"
"time"
"telesrv/internal/domain"
)
func TestClientTelemetryStoreIdempotencyRateLimitAndRetention(t *testing.T) {
ctx := context.Background()
store := NewClientTelemetryStore()
now := time.Unix(1_750_000_000, 0).UTC()
newEvent := func(subject int64, at time.Time) domain.ClientTelemetryEvent {
event, err := domain.NewClientTelemetryEvent(
71, domain.ClientTelemetryMessageDelivery,
domain.Peer{Type: domain.PeerTypeUser, ID: 72},
[]int64{subject}, map[string]any{"push": true}, at,
)
if err != nil {
t.Fatal(err)
}
return event
}
first := newEvent(1, now)
stored, created, err := store.CreateClientTelemetry(ctx, first)
if err != nil || !created || stored.ID <= 0 {
t.Fatalf("first=%+v created=%v err=%v", stored, created, err)
}
retry, created, err := store.CreateClientTelemetry(ctx, first)
if err != nil || created || retry.ID != stored.ID {
t.Fatalf("retry=%+v created=%v err=%v", retry, created, err)
}
for i := 1; i < domain.MaxClientTelemetryEventsPerHour; i++ {
if _, created, err := store.CreateClientTelemetry(
ctx, newEvent(int64(i+1), now),
); err != nil || !created {
t.Fatalf("create %d created=%v err=%v", i, created, err)
}
}
if got, created, err := store.CreateClientTelemetry(ctx, first); err != nil ||
created || got.ID != stored.ID {
t.Fatalf("retry at limit got=%+v created=%v err=%v", got, created, err)
}
if _, _, err := store.CreateClientTelemetry(
ctx, newEvent(domain.MaxClientTelemetryEventsPerHour+1, now),
); !errors.Is(err, domain.ErrClientTelemetryRateLimited) {
t.Fatalf("overflow err=%v", err)
}
deleted, err := store.DeleteExpiredClientTelemetry(
ctx, now.Add(time.Second), domain.MaxClientTelemetryEventsPerHour+1,
)
if err != nil || deleted != domain.MaxClientTelemetryEventsPerHour {
t.Fatalf("deleted=%d err=%v", deleted, err)
}
recreated, created, err := store.CreateClientTelemetry(ctx, first)
if err != nil || !created || recreated.ID == stored.ID {
t.Fatalf("recreated=%+v created=%v err=%v", recreated, created, err)
}
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,221 @@
package memory
import (
"context"
"testing"
"time"
"telesrv/internal/domain"
)
func TestModerationCaseLifecycleAndNewReportsDuringAction(t *testing.T) {
ctx := context.Background()
now := time.Now().UTC()
target := domain.Peer{Type: domain.PeerTypeUser, ID: 900}
store := NewModerationReportStore()
create := func(reporter int64, option string, at time.Time) domain.ModerationReport {
report, err := domain.NewModerationReport(domain.ModerationReportDraft{
ReporterUserID: reporter, Source: domain.ModerationSourceAccountPeer,
Target: target, Reason: domain.ModerationReasonFake,
Option: option,
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemPeer, Peer: target, ItemID: target.ID,
AuthorUserID: target.ID, EvidenceSchemaVersion: 1,
Evidence: []byte(`{"schema_version":1}`),
}},
CreatedAt: at,
})
if err != nil {
t.Fatal(err)
}
stored, created, err := store.CreateModerationReport(ctx, report)
if err != nil || !created {
t.Fatalf("create report created=%v err=%v", created, err)
}
return stored
}
create(101, "fake", now)
create(102, "fake:impersonation", now.Add(time.Second))
cases, err := store.ListModerationCases(ctx, domain.ModerationCaseFilter{Limit: 10})
if err != nil || len(cases) != 1 {
t.Fatalf("cases=%+v err=%v", cases, err)
}
item := cases[0]
if item.ReportCount != 2 || item.DistinctReporterCount != 2 ||
item.Version != 2 || item.Severity != domain.ModerationSeverityMedium {
t.Fatalf("case aggregate=%+v", item)
}
claimed, err := store.ClaimModerationCase(ctx, item.ID, item.Version, "reviewer", now.Add(2*time.Second))
if err != nil || claimed.Status != domain.ModerationCaseInReview {
t.Fatalf("claim=%+v err=%v", claimed, err)
}
decision, err := domain.NewModerationDecisionRequest(domain.ModerationDecisionRequest{
CaseID: item.ID, ExpectedVersion: claimed.Version, Actor: "reviewer",
Reason: "confirmed impersonation", CommandID: "decision-1",
Kind: domain.ModerationDecisionViolation,
Actions: []domain.ModerationActionDraft{{
Kind: domain.ModerationActionMarkFake, Payload: []byte(`{}`),
}},
CreatedAt: now.Add(3 * time.Second),
})
if err != nil {
t.Fatal(err)
}
detail, created, err := store.DecideModerationCase(ctx, decision)
if err != nil || !created ||
detail.Case.Status != domain.ModerationCaseActionPending ||
len(detail.Actions) != 1 {
t.Fatalf("decision detail=%+v created=%v err=%v", detail, created, err)
}
if _, created, err := store.DecideModerationCase(ctx, decision); err != nil || created {
t.Fatalf("decision retry created=%v err=%v", created, err)
}
// Once a decision is durable, later reports open a new case instead of
// mutating the evidence set under the pending action.
create(103, "fake:new-evidence", now.Add(4*time.Second))
cases, err = store.ListModerationCases(ctx, domain.ModerationCaseFilter{Limit: 10})
if err != nil || len(cases) != 2 {
t.Fatalf("cases after new evidence=%+v err=%v", cases, err)
}
actions, err := store.ClaimModerationActions(ctx, now.Add(5*time.Second), 10, time.Minute)
if err != nil || len(actions) != 1 {
t.Fatalf("claimed actions=%+v err=%v", actions, err)
}
if err := store.CompleteModerationAction(
ctx, actions[0].ID, actions[0].Attempts, true, "",
time.Time{}, now.Add(6*time.Second),
); err != nil {
t.Fatal(err)
}
resolved, found, err := store.GetModerationCase(ctx, item.ID)
if err != nil || !found || resolved.Case.Status != domain.ModerationCaseResolved {
t.Fatalf("resolved=%+v found=%v err=%v", resolved, found, err)
}
appeal, err := domain.NewModerationAppeal(
item.ID, target.ID, domain.ModerationCaseResolved,
"This is a mistake.", now.Add(7*time.Second),
)
if err != nil {
t.Fatal(err)
}
if _, created, err := store.CreateModerationAppeal(ctx, appeal); err != nil || !created {
t.Fatalf("appeal created=%v err=%v", created, err)
}
appealed, _, _ := store.GetModerationCase(ctx, item.ID)
if appealed.Case.Status != domain.ModerationCaseAppealReview ||
len(appealed.Appeals) != 1 {
t.Fatalf("appealed detail=%+v", appealed)
}
}
func TestModerationActionFailedCanBeRedrivenByNewDecision(t *testing.T) {
ctx := context.Background()
now := time.Unix(1_750_000_000, 0).UTC()
target := domain.Peer{Type: domain.PeerTypeUser, ID: 902}
store := NewModerationReportStore()
report, err := domain.NewModerationReport(domain.ModerationReportDraft{
ReporterUserID: 901, Source: domain.ModerationSourceAccountPeer,
Target: target, Reason: domain.ModerationReasonSpam, Option: "spam",
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemPeer, Peer: target, ItemID: target.ID,
AuthorUserID: target.ID, EvidenceSchemaVersion: 1,
Evidence: []byte(`{"schema_version":1}`),
}},
CreatedAt: now,
})
if err != nil {
t.Fatal(err)
}
if _, created, err := store.CreateModerationReport(ctx, report); err != nil || !created {
t.Fatalf("create report created=%v err=%v", created, err)
}
cases, err := store.ListModerationCases(ctx, domain.ModerationCaseFilter{Limit: 10})
if err != nil || len(cases) != 1 {
t.Fatalf("cases=%+v err=%v", cases, err)
}
claimed, err := store.ClaimModerationCase(
ctx, cases[0].ID, cases[0].Version, "reviewer", now.Add(time.Second),
)
if err != nil {
t.Fatal(err)
}
firstDecision, err := domain.NewModerationDecisionRequest(domain.ModerationDecisionRequest{
CaseID: claimed.ID, ExpectedVersion: claimed.Version,
Actor: "reviewer", Reason: "first command kept failing",
CommandID: "redrive-first", Kind: domain.ModerationDecisionViolation,
Actions: []domain.ModerationActionDraft{{
Kind: domain.ModerationActionMarkScam, Payload: []byte(`{}`),
}},
CreatedAt: now.Add(2 * time.Second),
})
if err != nil {
t.Fatal(err)
}
if _, created, err := store.DecideModerationCase(ctx, firstDecision); err != nil || !created {
t.Fatalf("first decision created=%v err=%v", created, err)
}
for attempt := 1; attempt <= domain.MaxModerationActionAttempts; attempt++ {
at := now.Add(time.Duration(attempt+2) * time.Second)
actions, err := store.ClaimModerationActions(ctx, at, 10, time.Second)
if err != nil || len(actions) != 1 {
t.Fatalf("attempt %d actions=%+v err=%v", attempt, actions, err)
}
if err := store.CompleteModerationAction(
ctx, actions[0].ID, actions[0].Attempts, false, "downstream unavailable",
at.Add(time.Millisecond), at,
); err != nil {
t.Fatalf("attempt %d: %v", attempt, err)
}
}
failed, found, err := store.GetModerationCase(ctx, claimed.ID)
if err != nil || !found || failed.Case.Status != domain.ModerationCaseActionFailed ||
len(failed.Actions) != 1 ||
failed.Actions[0].Status != domain.ModerationActionFailed {
t.Fatalf("failed=%+v found=%v err=%v", failed, found, err)
}
redrive, err := domain.NewModerationDecisionRequest(domain.ModerationDecisionRequest{
CaseID: claimed.ID, ExpectedVersion: failed.Case.Version,
Actor: "reviewer", Reason: "redrive after dependency recovery",
CommandID: "redrive-second", Kind: domain.ModerationDecisionViolation,
Actions: []domain.ModerationActionDraft{{
Kind: domain.ModerationActionMarkScam, Payload: []byte(`{}`),
}},
CreatedAt: now.Add(time.Minute),
})
if err != nil {
t.Fatal(err)
}
pending, created, err := store.DecideModerationCase(ctx, redrive)
if err != nil || !created ||
pending.Case.Status != domain.ModerationCaseActionPending ||
len(pending.Actions) != 2 {
t.Fatalf("pending=%+v created=%v err=%v", pending, created, err)
}
actions, err := store.ClaimModerationActions(ctx, now.Add(2*time.Minute), 10, time.Second)
if err != nil || len(actions) != 1 || actions[0].DecisionID == failed.Actions[0].DecisionID {
t.Fatalf("redrive actions=%+v err=%v", actions, err)
}
if err := store.CompleteModerationAction(
ctx, actions[0].ID, actions[0].Attempts, true, "",
time.Time{}, now.Add(2*time.Minute+time.Second),
); err != nil {
t.Fatal(err)
}
resolved, found, err := store.GetModerationCase(ctx, claimed.ID)
if err != nil || !found || resolved.Case.Status != domain.ModerationCaseResolved {
t.Fatalf("resolved=%+v found=%v err=%v", resolved, found, err)
}
var failedCount, succeededCount int
for _, action := range resolved.Actions {
switch action.Status {
case domain.ModerationActionFailed:
failedCount++
case domain.ModerationActionSucceeded:
succeededCount++
}
}
if failedCount != 1 || succeededCount != 1 {
t.Fatalf("action history failed=%d succeeded=%d", failedCount, succeededCount)
}
}

View file

@ -0,0 +1,92 @@
package memory
import (
"context"
"testing"
"time"
"telesrv/internal/domain"
)
func TestModerationReportStoreIdempotencyAndCopyIsolation(t *testing.T) {
report, err := domain.NewModerationReport(domain.ModerationReportDraft{
ReporterUserID: 11, Source: domain.ModerationSourceMessages,
Target: domain.Peer{Type: domain.PeerTypeUser, ID: 22},
Reason: domain.ModerationReasonSpam, Option: "v1/spam",
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemMessage,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: 22},
ItemID: 5, AuthorUserID: 22, EvidenceSchemaVersion: 1,
Evidence: []byte(`{"message":"spam"}`),
}},
CreatedAt: time.Now().UTC(),
})
if err != nil {
t.Fatal(err)
}
store := NewModerationReportStore()
first, created, err := store.CreateModerationReport(context.Background(), report)
if err != nil || !created || first.ID <= 0 {
t.Fatalf("first = %+v created=%v err=%v", first, created, err)
}
first.Items[0].Evidence[0] = '['
retry, created, err := store.CreateModerationReport(context.Background(), report)
if err != nil || created || retry.ID != first.ID {
t.Fatalf("retry = %+v created=%v err=%v", retry, created, err)
}
if retry.Items[0].Evidence[0] != '{' {
t.Fatalf("caller mutation changed stored evidence: %s", retry.Items[0].Evidence)
}
}
func TestModerationReportStoreRateLimitDoesNotChargeIdempotentRetry(t *testing.T) {
ctx := context.Background()
store := NewModerationReportStore()
now := time.Unix(1_750_000_000, 0).UTC()
var first domain.ModerationReport
for i := 0; i < domain.MaxModerationReportsPerHour; i++ {
report, err := domain.NewModerationReport(domain.ModerationReportDraft{
ReporterUserID: 71, Source: domain.ModerationSourceMessages,
Target: domain.Peer{Type: domain.PeerTypeUser, ID: 72},
Reason: domain.ModerationReasonSpam, Option: "spam",
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemMessage,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: 72},
ItemID: int64(i + 1), AuthorUserID: 72,
EvidenceSchemaVersion: 1,
Evidence: []byte(`{"message":"spam"}`),
}},
CreatedAt: now,
})
if err != nil {
t.Fatal(err)
}
if _, created, err := store.CreateModerationReport(ctx, report); err != nil || !created {
t.Fatalf("create %d: created=%v err=%v", i, created, err)
}
if i == 0 {
first = report
}
}
if got, created, err := store.CreateModerationReport(ctx, first); err != nil || created || got.ID == 0 {
t.Fatalf("retry after limit: got=%+v created=%v err=%v", got, created, err)
}
overflow, err := domain.NewModerationReport(domain.ModerationReportDraft{
ReporterUserID: 71, Source: domain.ModerationSourceMessages,
Target: domain.Peer{Type: domain.PeerTypeUser, ID: 72},
Reason: domain.ModerationReasonSpam, Option: "spam",
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemMessage,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: 72},
ItemID: 999, AuthorUserID: 72, EvidenceSchemaVersion: 1,
Evidence: []byte(`{"message":"overflow"}`),
}},
CreatedAt: now,
})
if err != nil {
t.Fatal(err)
}
if _, _, err := store.CreateModerationReport(ctx, overflow); err != domain.ErrModerationRateLimited {
t.Fatalf("overflow err=%v, want ErrModerationRateLimited", err)
}
}

View file

@ -160,6 +160,18 @@ func (s *PasswordStore) GetAccountSettings(_ context.Context, userID int64) (dom
return settings, ok, nil // AccountSettings 全是值类型,无需深拷贝
}
func (s *PasswordStore) GetAccountSettingsBatch(_ context.Context, userIDs []int64) (map[int64]domain.AccountSettings, error) {
out := make(map[int64]domain.AccountSettings, len(userIDs))
s.mu.RLock()
for _, userID := range userIDs {
if settings, ok := s.accountSettings[userID]; ok {
out[userID] = settings
}
}
s.mu.RUnlock()
return out, nil
}
func (s *PasswordStore) SaveAccountSettings(_ context.Context, userID int64, settings domain.AccountSettings) error {
s.mu.Lock()
s.accountSettings[userID] = settings

View file

@ -105,9 +105,20 @@ func cloneUpdateEvent(event domain.UpdateEvent) domain.UpdateEvent {
event.Reaction = cloneUpdateReaction(event.Reaction)
event.QuickReplies = cloneUpdateQuickReplies(event.QuickReplies)
event.QuickReplyMessage = cloneUpdateQuickReplyMessage(event.QuickReplyMessage)
event.Privacy.Rules = clonePrivacyRuleSlice(event.Privacy.Rules)
return event
}
func clonePrivacyRuleSlice(in []domain.PrivacyRule) []domain.PrivacyRule {
out := make([]domain.PrivacyRule, len(in))
for i := range in {
out[i] = in[i]
out[i].UserIDs = append([]int64(nil), in[i].UserIDs...)
out[i].ChatIDs = append([]int64(nil), in[i].ChatIDs...)
}
return out
}
func cloneUpdateStory(story domain.Story) domain.Story {
story.Entities = append([]domain.MessageEntity(nil), story.Entities...)
story.Views.Reactions = append([]domain.ChannelMessageReactionCount(nil), story.Views.Reactions...)

View file

@ -0,0 +1,64 @@
package store
import (
"context"
"time"
"telesrv/internal/domain"
)
// ModerationReportStore atomically persists an immutable report, all evidence
// items and media holds. A retry with the same fingerprint returns the original
// report and created=false; implementations must never create partial items.
type ModerationReportStore interface {
CreateModerationReport(ctx context.Context, report domain.ModerationReport) (stored domain.ModerationReport, created bool, err error)
GetModerationReport(ctx context.Context, reportID int64) (domain.ModerationReport, bool, error)
}
type ModerationEvidenceRegistryStore interface {
CreateSponsoredMessageImpression(ctx context.Context, impression domain.SponsoredMessageImpression) (domain.SponsoredMessageImpression, bool, error)
GetSponsoredMessageImpression(ctx context.Context, userID int64, randomIDHash [32]byte, now time.Time) (domain.SponsoredMessageImpression, bool, error)
CreateSponsoredModerationReport(ctx context.Context, impressionID int64, report domain.ModerationReport) (domain.ModerationReport, bool, error)
CreateChannelAntiSpamDecision(ctx context.Context, decision domain.ChannelAntiSpamDecision) (domain.ChannelAntiSpamDecision, bool, error)
GetChannelAntiSpamDecision(ctx context.Context, channelID int64, messageID int) (domain.ChannelAntiSpamDecision, bool, error)
CreateAntiSpamFalsePositiveReport(ctx context.Context, decisionID int64, report domain.ModerationReport) (domain.ModerationReport, bool, error)
DeleteExpiredSponsoredMessageImpressions(ctx context.Context, olderThan time.Time, limit int) (int, error)
}
// LegacyEphemeralReport is an immutable row produced before ephemeral reports
// joined the unified moderation pipeline.
type LegacyEphemeralReport struct {
ID int64
Report domain.EphemeralAbuseReport
}
// LegacyEphemeralReportReader exposes only still-unmapped rows. It exists for
// the startup migration and must not be injected into RPC handlers.
type LegacyEphemeralReportReader interface {
ListUnmigratedEphemeralReports(ctx context.Context, limit int) ([]LegacyEphemeralReport, error)
}
// LegacyEphemeralReportImporter atomically inserts the unified report and its
// provenance mapping. Historical imports bypass submission rate limits but
// retain normal report validation and fingerprint idempotency.
type LegacyEphemeralReportImporter interface {
ImportLegacyEphemeralReport(ctx context.Context, legacyReportID int64, report domain.ModerationReport) (stored domain.ModerationReport, created bool, err error)
}
type ModerationCaseStore interface {
ListModerationCases(ctx context.Context, filter domain.ModerationCaseFilter) ([]domain.ModerationCase, error)
GetModerationCase(ctx context.Context, caseID int64) (domain.ModerationCaseDetail, bool, error)
ClaimModerationCase(ctx context.Context, caseID, expectedVersion int64, actor string, now time.Time) (domain.ModerationCase, error)
DecideModerationCase(ctx context.Context, request domain.ModerationDecisionRequest) (domain.ModerationCaseDetail, bool, error)
ReviewModerationAppeal(ctx context.Context, request domain.ModerationDecisionRequest) (domain.ModerationCaseDetail, bool, error)
CreateModerationAppeal(ctx context.Context, appeal domain.ModerationAppeal) (domain.ModerationAppeal, bool, error)
GetModerationAppeal(ctx context.Context, appealID int64) (domain.ModerationAppeal, bool, error)
IssueModerationAppealLink(ctx context.Context, link domain.ModerationAppealLink) (domain.ModerationAppealLink, error)
GetModerationAppealLink(ctx context.Context, tokenHash [32]byte, now time.Time) (domain.ModerationAppealLink, bool, error)
SubmitModerationAppealByLink(ctx context.Context, tokenHash [32]byte, text string, now time.Time) (domain.ModerationAppeal, bool, error)
DeleteExpiredModerationAppealLinks(ctx context.Context, olderThan time.Time, limit int) (int, error)
ClaimModerationActions(ctx context.Context, now time.Time, limit int, lease time.Duration) ([]domain.ModerationAction, error)
IsModerationActionCurrent(ctx context.Context, action domain.ModerationAction) (bool, error)
SupersedeModerationAction(ctx context.Context, actionID int64, expectedAttempts int, now time.Time) error
CompleteModerationAction(ctx context.Context, actionID int64, expectedAttempts int, succeeded bool, errorText string, retryAt, now time.Time) error
}

View file

@ -261,6 +261,41 @@ WHERE user_id = $1`, userID)
return settings, true, nil
}
func (s *PasswordStore) GetAccountSettingsBatch(ctx context.Context, userIDs []int64) (map[int64]domain.AccountSettings, error) {
out := make(map[int64]domain.AccountSettings, len(userIDs))
if len(userIDs) == 0 {
return out, nil
}
rows, err := s.db.Query(ctx, `
SELECT user_id, archive_and_mute_new_noncontact_peers, keep_archived_unmuted, keep_archived_folders,
hide_read_marks, new_noncontact_peers_require_premium, display_gifts_button,
noncontact_peers_paid_stars, account_ttl_days, sensitive_content_enabled, contact_signup_silent
FROM account_settings
WHERE user_id = ANY($1::bigint[])`, userIDs)
if err != nil {
return nil, fmt.Errorf("get account settings batch: %w", err)
}
defer rows.Close()
for rows.Next() {
var userID int64
settings := domain.DefaultAccountSettings()
gp := &settings.GlobalPrivacy
if err := rows.Scan(
&userID,
&gp.ArchiveAndMuteNewNoncontactPeers, &gp.KeepArchivedUnmuted, &gp.KeepArchivedFolders,
&gp.HideReadMarks, &gp.NewNoncontactPeersRequirePremium, &gp.DisplayGiftsButton,
&gp.NoncontactPeersPaidStars, &settings.AccountTTLDays, &settings.SensitiveContentEnabled, &settings.ContactSignUpSilent,
); err != nil {
return nil, fmt.Errorf("scan account settings batch: %w", err)
}
out[userID] = settings
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate account settings batch: %w", err)
}
return out, nil
}
func (s *PasswordStore) SaveAccountSettings(ctx context.Context, userID int64, settings domain.AccountSettings) error {
gp := settings.GlobalPrivacy
if _, err := s.db.Exec(ctx, `

View file

@ -0,0 +1,152 @@
package postgres
import (
"context"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"telesrv/internal/domain"
"telesrv/internal/store/postgres/sqlcgen"
)
type AuthDeliveryReportStore struct {
db sqlcgen.DBTX
}
func NewAuthDeliveryReportStore(db sqlcgen.DBTX) *AuthDeliveryReportStore {
return &AuthDeliveryReportStore{db: db}
}
func (s *AuthDeliveryReportStore) CreateAuthDeliveryReport(ctx context.Context, report domain.AuthDeliveryReport) (domain.AuthDeliveryReport, bool, error) {
if s == nil || s.db == nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("auth delivery report store is not configured")
}
if err := report.Validate(); err != nil {
return domain.AuthDeliveryReport{}, false, err
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("auth delivery report store requires transaction-capable postgres handle")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("begin auth delivery report: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
if _, err := tx.Exec(ctx, `
SELECT pg_advisory_xact_lock(
hashtextextended('auth-delivery:' || encode($1::bytea, 'hex'), 0)
)`, report.AuthKeyID[:]); err != nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("lock auth delivery reporter: %w", err)
}
existing, found, err := getAuthDeliveryReportByFingerprint(ctx, tx, report.AuthKeyID, report.Fingerprint)
if err != nil {
return domain.AuthDeliveryReport{}, false, err
}
if found {
return existing, false, nil
}
var hourly, phoneDaily int
if err := tx.QueryRow(ctx, `
SELECT
count(*) FILTER (
WHERE auth_key_id = $1 AND created_at >= $3::timestamptz - interval '1 hour'
),
count(*) FILTER (
WHERE phone_hash = $2 AND created_at >= $3::timestamptz - interval '24 hours'
)
FROM auth_delivery_reports
WHERE created_at <= $3::timestamptz
AND (auth_key_id = $1 OR phone_hash = $2)`,
report.AuthKeyID[:], report.PhoneHash[:], report.CreatedAt,
).Scan(&hourly, &phoneDaily); err != nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("count auth delivery reports: %w", err)
}
if hourly >= domain.MaxAuthDeliveryReportsPerHour ||
phoneDaily >= domain.MaxAuthDeliveryReportsPerPhoneDay {
return domain.AuthDeliveryReport{}, false, domain.ErrAuthDeliveryRateLimited
}
err = tx.QueryRow(ctx, `
INSERT INTO auth_delivery_reports (
auth_key_id, session_id, client_type, phone_hash, code_hash,
issued_user_id, delivery_id, channel, mnc, fingerprint, created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
RETURNING id`,
report.AuthKeyID[:], report.SessionID, report.ClientType,
report.PhoneHash[:], report.CodeHash[:], report.IssuedUserID,
report.DeliveryID, string(report.Channel), report.MNC,
report.Fingerprint[:], report.CreatedAt,
).Scan(&report.ID)
if err != nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("insert auth delivery report: %w", err)
}
if err := tx.Commit(ctx); err != nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("commit auth delivery report: %w", err)
}
return report, true, nil
}
func getAuthDeliveryReportByFingerprint(ctx context.Context, db sqlcgen.DBTX, authKeyID [8]byte, fingerprint [32]byte) (domain.AuthDeliveryReport, bool, error) {
var report domain.AuthDeliveryReport
var storedAuthKey, phoneHash, codeHash, storedFingerprint []byte
var channel string
err := db.QueryRow(ctx, `
SELECT id, auth_key_id, session_id, client_type, phone_hash, code_hash,
issued_user_id, delivery_id, channel, mnc, fingerprint, created_at
FROM auth_delivery_reports
WHERE auth_key_id = $1 AND fingerprint = $2`,
authKeyID[:], fingerprint[:],
).Scan(
&report.ID, &storedAuthKey, &report.SessionID, &report.ClientType,
&phoneHash, &codeHash, &report.IssuedUserID, &report.DeliveryID,
&channel, &report.MNC, &storedFingerprint, &report.CreatedAt,
)
if errors.Is(err, pgx.ErrNoRows) {
return domain.AuthDeliveryReport{}, false, nil
}
if err != nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("get auth delivery report: %w", err)
}
if len(storedAuthKey) != len(report.AuthKeyID) ||
len(phoneHash) != len(report.PhoneHash) ||
len(codeHash) != len(report.CodeHash) ||
len(storedFingerprint) != len(report.Fingerprint) {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("get auth delivery report: invalid hash shape")
}
copy(report.AuthKeyID[:], storedAuthKey)
copy(report.PhoneHash[:], phoneHash)
copy(report.CodeHash[:], codeHash)
copy(report.Fingerprint[:], storedFingerprint)
report.Channel = domain.AuthCodeDeliveryKind(channel)
if err := report.Validate(); err != nil {
return domain.AuthDeliveryReport{}, false, fmt.Errorf("validate auth delivery report: %w", err)
}
return report, true, nil
}
func (s *AuthDeliveryReportStore) DeleteExpiredAuthDeliveryReports(ctx context.Context, olderThan time.Time, limit int) (int, error) {
if s == nil || s.db == nil {
return 0, fmt.Errorf("auth delivery report store is not configured")
}
if olderThan.IsZero() || limit <= 0 || limit > 10000 {
return 0, domain.ErrAuthDeliveryReportInvalid
}
tag, err := s.db.Exec(ctx, `
WITH doomed AS (
SELECT id
FROM auth_delivery_reports
WHERE created_at < $1
ORDER BY created_at, id
LIMIT $2
)
DELETE FROM auth_delivery_reports r
USING doomed d
WHERE r.id = d.id`, olderThan, limit)
if err != nil {
return 0, fmt.Errorf("delete expired auth delivery reports: %w", err)
}
return int(tag.RowsAffected()), nil
}

View file

@ -0,0 +1,53 @@
package postgres
import (
"context"
"testing"
"time"
"telesrv/internal/domain"
)
func TestAuthDeliveryReportPostgresIsIdempotentAndRetainedSeparately(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
createdAt := time.Unix(123_456, 0).UTC()
report, err := domain.NewAuthDeliveryReport(
[8]byte{1, 2, 3, 4, 5, 6, 7, byte(time.Now().UnixNano())},
time.Now().UnixNano(), "tdesktop", "15550004444",
"phone-code-hash", 77, "delivery-77",
domain.AuthCodeDeliverySMS, "46000", createdAt,
)
if err != nil {
t.Fatal(err)
}
store := NewAuthDeliveryReportStore(pool)
stored, created, err := store.CreateAuthDeliveryReport(ctx, report)
if err != nil || !created || stored.ID <= 0 {
t.Fatalf("stored=%+v created=%v err=%v", stored, created, err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM auth_delivery_reports WHERE id = $1", stored.ID)
})
retry, created, err := store.CreateAuthDeliveryReport(ctx, report)
if err != nil || created || retry.ID != stored.ID ||
retry.PhoneHash != stored.PhoneHash || retry.CodeHash != stored.CodeHash {
t.Fatalf("retry=%+v created=%v err=%v", retry, created, err)
}
deleted, err := store.DeleteExpiredAuthDeliveryReports(
ctx, createdAt.Add(time.Second), 10,
)
if err != nil || deleted < 1 {
t.Fatalf("deleted=%d err=%v", deleted, err)
}
var moderationRows int
if err := pool.QueryRow(ctx, `
SELECT count(*)
FROM moderation_reports
WHERE reporter_user_id = $1`, report.IssuedUserID).Scan(&moderationRows); err != nil {
t.Fatal(err)
}
if moderationRows != 0 {
t.Fatalf("auth diagnostic leaked into moderation reports: %d", moderationRows)
}
}

View file

@ -60,6 +60,68 @@ func (s *ChannelStore) DeleteChannelMessages(ctx context.Context, req domain.Del
return domain.DeleteChannelMessagesResult{Channel: channel, Event: event, DeletedIDs: deleted, Recipients: recipients, DiscussionDeletes: cascades}, nil
}
func (s *ChannelStore) ModerationDeleteChannelMessages(ctx context.Context, channelID int64, ids []int, date int) (domain.DeleteChannelMessagesResult, error) {
if channelID <= 0 || len(ids) == 0 || len(ids) > domain.MaxDeleteMessageIDs {
return domain.DeleteChannelMessagesResult{}, domain.ErrChannelInvalid
}
if date == 0 {
date = nowUnix()
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.DeleteChannelMessagesResult{}, fmt.Errorf("moderation delete channel messages: db does not support transactions")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.DeleteChannelMessagesResult{}, fmt.Errorf("begin moderation delete channel messages: %w", err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback(ctx)
}
}()
channel, err := getChannelByID(ctx, tx, channelID)
if err != nil || channel.Deleted {
if err != nil {
return domain.DeleteChannelMessagesResult{}, err
}
return domain.DeleteChannelMessagesResult{}, domain.ErrChannelInvalid
}
refs, err := s.discussionRefsForMessages(ctx, tx, channel.ID, ids)
if err != nil {
return domain.DeleteChannelMessagesResult{}, err
}
systemMember := domain.ChannelMember{
ChannelID: channel.ID, UserID: domain.OfficialSystemUserID,
Role: domain.ChannelRoleCreator, Status: domain.ChannelMemberActive,
}
deleted, event, channel, err := s.deleteChannelMessagesTx(
ctx, tx, channel, systemMember, ids, domain.OfficialSystemUserID, date,
)
if err != nil {
return domain.DeleteChannelMessagesResult{}, err
}
cascades, err := s.cascadeDeleteDiscussionRootsTx(
ctx, tx, refs, deleted, domain.OfficialSystemUserID, date,
)
if err != nil {
return domain.DeleteChannelMessagesResult{}, err
}
if err := tx.Commit(ctx); err != nil {
return domain.DeleteChannelMessagesResult{}, fmt.Errorf("commit moderation delete channel messages: %w", err)
}
committed = true
recipients, _ := s.ListActiveChannelMemberIDs(ctx, 0, channel.ID, 0)
for i := range cascades {
cascades[i].Recipients, _ = s.ListActiveChannelMemberIDs(ctx, 0, cascades[i].Channel.ID, 0)
}
return domain.DeleteChannelMessagesResult{
Channel: channel, Event: event, DeletedIDs: deleted,
Recipients: recipients, DiscussionDeletes: cascades,
}, nil
}
// discussionRefsForMessages 取待删消息携带的讨论组转发根引用。
func (s *ChannelStore) discussionRefsForMessages(ctx context.Context, tx pgx.Tx, channelID int64, ids []int) (map[int]domain.ChannelDiscussionRef, error) {
id32, _, err := validUniqueChannelMessageIDs(ids)

View file

@ -612,3 +612,48 @@ LIMIT $`+fmt.Sprint(len(args)), args...)
NextOffset: next,
}, nil
}
func (s *ChannelStore) FindChannelMessageReaction(ctx context.Context, req domain.ChannelMessageReactionLookupRequest) (domain.ChannelMessageReactionLookup, bool, error) {
if req.ViewerUserID == 0 || req.ChannelID == 0 || req.MessageID <= 0 ||
req.MessageID > domain.MaxMessageBoxID || req.ReactorUserID == 0 {
return domain.ChannelMessageReactionLookup{}, false, domain.ErrChannelInvalid
}
channel, member, err := s.getChannelForMember(ctx, s.db, req.ViewerUserID, req.ChannelID)
if err != nil {
return domain.ChannelMessageReactionLookup{}, false, err
}
message, err := s.getChannelMessage(ctx, s.db, req.ChannelID, req.MessageID)
if err != nil {
return domain.ChannelMessageReactionLookup{}, false, err
}
if message.Deleted || message.ID <= member.AvailableMinID {
return domain.ChannelMessageReactionLookup{}, false, domain.ErrMessageIDInvalid
}
rows, err := s.db.Query(ctx, `
SELECT channel_id, message_id, reacted_user_id, sender_user_id,
reaction_type, reaction_value, big, unread, chosen_order, reaction_date
FROM channel_message_reactions
WHERE channel_id = $1 AND message_id = $2 AND reacted_user_id = $3
ORDER BY chosen_order, reaction_type, reaction_value
LIMIT $4`,
req.ChannelID, req.MessageID, req.ReactorUserID,
domain.MaxChannelMessageReactionsPerUser)
if err != nil {
return domain.ChannelMessageReactionLookup{}, false, fmt.Errorf("find channel message reaction: %w", err)
}
defer rows.Close()
reactions := make([]domain.ChannelMessagePeerReaction, 0, domain.MaxChannelMessageReactionsPerUser)
for rows.Next() {
reaction, err := scanChannelMessagePeerReaction(rows, req.ViewerUserID)
if err != nil {
return domain.ChannelMessageReactionLookup{}, false, err
}
reactions = append(reactions, reaction)
}
if err := rows.Err(); err != nil {
return domain.ChannelMessageReactionLookup{}, false, err
}
return domain.ChannelMessageReactionLookup{
Channel: channel, Message: message, Reactions: reactions,
}, len(reactions) > 0, nil
}

View file

@ -0,0 +1,149 @@
package postgres
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"telesrv/internal/domain"
"telesrv/internal/store/postgres/sqlcgen"
)
type ClientTelemetryStore struct {
db sqlcgen.DBTX
}
func NewClientTelemetryStore(db sqlcgen.DBTX) *ClientTelemetryStore {
return &ClientTelemetryStore{db: db}
}
func (s *ClientTelemetryStore) CreateClientTelemetry(ctx context.Context, event domain.ClientTelemetryEvent) (domain.ClientTelemetryEvent, bool, error) {
if s == nil || s.db == nil {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("client telemetry store is not configured")
}
if err := event.Validate(); err != nil || event.ID != 0 {
return domain.ClientTelemetryEvent{}, false, domain.ErrClientTelemetryInvalid
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("client telemetry store requires transaction-capable postgres handle")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("begin client telemetry: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
if _, err := tx.Exec(ctx, `
SELECT pg_advisory_xact_lock(
hashtextextended('client-telemetry:' || $1::bigint::text, 0)
)`, event.UserID); err != nil {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("lock client telemetry user: %w", err)
}
existing, found, err := getClientTelemetryByFingerprint(
ctx, tx, event.UserID, event.Fingerprint,
)
if err != nil {
return domain.ClientTelemetryEvent{}, false, err
}
if found {
return existing, false, nil
}
var hourly, daily int
if err := tx.QueryRow(ctx, `
SELECT
count(*) FILTER (WHERE created_at >= $2::timestamptz - interval '1 hour'),
count(*) FILTER (WHERE created_at >= $2::timestamptz - interval '24 hours')
FROM client_telemetry_events
WHERE user_id = $1 AND created_at <= $2::timestamptz`,
event.UserID, event.CreatedAt,
).Scan(&hourly, &daily); err != nil {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("count client telemetry: %w", err)
}
if hourly >= domain.MaxClientTelemetryEventsPerHour ||
daily >= domain.MaxClientTelemetryEventsPerDay {
return domain.ClientTelemetryEvent{}, false, domain.ErrClientTelemetryRateLimited
}
if err := tx.QueryRow(ctx, `
INSERT INTO client_telemetry_events (
user_id, kind, peer_type, peer_id, subject_ids, payload,
fingerprint, created_at
) VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8)
RETURNING id`,
event.UserID, string(event.Kind), string(event.Peer.Type),
event.Peer.ID, event.SubjectIDs, []byte(event.Payload),
event.Fingerprint[:], event.CreatedAt,
).Scan(&event.ID); err != nil {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("insert client telemetry: %w", err)
}
if err := tx.Commit(ctx); err != nil {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("commit client telemetry: %w", err)
}
return event, true, nil
}
func (s *ClientTelemetryStore) DeleteExpiredClientTelemetry(ctx context.Context, olderThan time.Time, limit int) (int, error) {
if s == nil || s.db == nil {
return 0, fmt.Errorf("client telemetry store is not configured")
}
if olderThan.IsZero() || limit <= 0 || limit > 10000 {
return 0, domain.ErrClientTelemetryInvalid
}
tag, err := s.db.Exec(ctx, `
WITH doomed AS (
SELECT id
FROM client_telemetry_events
WHERE created_at < $1
ORDER BY created_at, id
LIMIT $2
)
DELETE FROM client_telemetry_events e
USING doomed d
WHERE e.id = d.id`, olderThan, limit)
if err != nil {
return 0, fmt.Errorf("delete expired client telemetry: %w", err)
}
return int(tag.RowsAffected()), nil
}
func getClientTelemetryByFingerprint(ctx context.Context, db sqlcgen.DBTX, userID int64, fingerprint [32]byte) (domain.ClientTelemetryEvent, bool, error) {
var event domain.ClientTelemetryEvent
var kind, peerType string
var payload, storedFingerprint []byte
if err := db.QueryRow(ctx, `
SELECT id, user_id, kind, peer_type, peer_id, subject_ids, payload,
fingerprint, created_at
FROM client_telemetry_events
WHERE user_id = $1 AND fingerprint = $2`,
userID, fingerprint[:],
).Scan(
&event.ID, &event.UserID, &kind, &peerType, &event.Peer.ID,
&event.SubjectIDs, &payload, &storedFingerprint, &event.CreatedAt,
); errors.Is(err, pgx.ErrNoRows) {
return domain.ClientTelemetryEvent{}, false, nil
} else if err != nil {
return domain.ClientTelemetryEvent{}, false, fmt.Errorf("get client telemetry: %w", err)
}
event.Kind = domain.ClientTelemetryKind(kind)
event.Peer.Type = domain.PeerType(peerType)
var canonicalPayload map[string]any
if err := json.Unmarshal(payload, &canonicalPayload); err != nil || canonicalPayload == nil {
return domain.ClientTelemetryEvent{}, false, domain.ErrClientTelemetryInvalid
}
canonicalRaw, marshalErr := json.Marshal(canonicalPayload)
if marshalErr != nil {
return domain.ClientTelemetryEvent{}, false, domain.ErrClientTelemetryInvalid
}
event.Payload = canonicalRaw
if len(storedFingerprint) != len(event.Fingerprint) {
return domain.ClientTelemetryEvent{}, false, domain.ErrClientTelemetryInvalid
}
copy(event.Fingerprint[:], storedFingerprint)
if err := event.Validate(); err != nil {
return domain.ClientTelemetryEvent{}, false, err
}
return event, true, nil
}

View file

@ -1,11 +1,13 @@
package postgres
import (
"bytes"
"context"
"encoding/json"
"fmt"
"telesrv/internal/domain"
"telesrv/internal/store"
"telesrv/internal/store/postgres/sqlcgen"
)
@ -47,3 +49,69 @@ ON CONFLICT (
}
return tag.RowsAffected() == 1, nil
}
func (s *EphemeralReportStore) ListUnmigratedEphemeralReports(ctx context.Context, limit int) ([]store.LegacyEphemeralReport, error) {
if s == nil || s.db == nil {
return nil, fmt.Errorf("ephemeral report store is not configured")
}
if limit <= 0 || limit > 1000 {
return nil, fmt.Errorf("legacy ephemeral report batch limit out of range")
}
rows, err := s.db.Query(ctx, `
SELECT r.id, r.reporter_user_id, r.channel_id, r.ephemeral_message_id,
r.sender_user_id, r.receiver_user_id, r.report_option,
r.report_comment, r.comment_hash, r.payload_hash, r.evidence,
r.created_at
FROM ephemeral_abuse_reports r
LEFT JOIN moderation_legacy_ephemeral_migrations m
ON m.legacy_report_id = r.id
WHERE m.legacy_report_id IS NULL
ORDER BY r.id
LIMIT $1`, limit)
if err != nil {
return nil, fmt.Errorf("list unmigrated ephemeral reports: %w", err)
}
defer rows.Close()
out := make([]store.LegacyEphemeralReport, 0, limit)
for rows.Next() {
var (
legacy store.LegacyEphemeralReport
channelID, senderUserID, receiverUserID int64
messageID int
commentHash, payloadHash, evidenceRaw []byte
)
if err := rows.Scan(
&legacy.ID, &legacy.Report.ReporterUserID, &channelID,
&messageID, &senderUserID, &receiverUserID,
&legacy.Report.Option, &legacy.Report.Comment, &commentHash,
&payloadHash, &evidenceRaw, &legacy.Report.CreatedAt,
); err != nil {
return nil, fmt.Errorf("scan legacy ephemeral report: %w", err)
}
if legacy.ID <= 0 || len(commentHash) != len(legacy.Report.CommentHash) ||
len(payloadHash) != len(legacy.Report.Evidence.PayloadHash) {
return nil, fmt.Errorf("legacy ephemeral report %d has invalid persisted identity", legacy.ID)
}
copy(legacy.Report.CommentHash[:], commentHash)
if err := json.Unmarshal(evidenceRaw, &legacy.Report.Evidence); err != nil {
return nil, fmt.Errorf("decode legacy ephemeral report %d evidence: %w", legacy.ID, err)
}
if legacy.Report.Evidence.Peer.Type != domain.PeerTypeChannel ||
legacy.Report.Evidence.Peer.ID != channelID ||
legacy.Report.Evidence.MessageID != messageID ||
legacy.Report.Evidence.SenderUserID != senderUserID ||
legacy.Report.Evidence.ReceiverUserID != receiverUserID ||
!bytes.Equal(legacy.Report.Evidence.PayloadHash[:], payloadHash) {
return nil, fmt.Errorf("legacy ephemeral report %d evidence disagrees with indexed columns", legacy.ID)
}
if err := legacy.Report.Validate(); err != nil {
return nil, fmt.Errorf("validate legacy ephemeral report %d: %w", legacy.ID, err)
}
out = append(out, legacy)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate legacy ephemeral reports: %w", err)
}
return out, nil
}

View file

@ -0,0 +1,441 @@
package postgres
import (
"context"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
"telesrv/internal/domain"
"telesrv/internal/store/postgres/sqlcgen"
)
type ModerationReportStore struct {
db sqlcgen.DBTX
}
func NewModerationReportStore(db sqlcgen.DBTX) *ModerationReportStore {
return &ModerationReportStore{db: db}
}
func (s *ModerationReportStore) CreateModerationReport(ctx context.Context, report domain.ModerationReport) (domain.ModerationReport, bool, error) {
if s == nil || s.db == nil {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store is not configured")
}
if err := report.Validate(); err != nil {
return domain.ModerationReport{}, false, err
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store requires transaction-capable postgres handle")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("begin moderation report: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
stored, created, err := createModerationReportTx(ctx, tx, report)
if err != nil {
return domain.ModerationReport{}, false, err
}
if err := tx.Commit(ctx); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("commit moderation report: %w", err)
}
return stored, created, nil
}
func createModerationReportTx(ctx context.Context, tx pgx.Tx, report domain.ModerationReport) (domain.ModerationReport, bool, error) {
var (
reportID int64
err error
)
if _, err := tx.Exec(ctx, `
SELECT pg_advisory_xact_lock(
hashtextextended('moderation-report:' || $1::bigint::text, 0)
)`, report.ReporterUserID); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("lock moderation reporter: %w", err)
}
err = tx.QueryRow(ctx, `
SELECT id
FROM moderation_reports
WHERE reporter_user_id = $1 AND fingerprint = $2`,
report.ReporterUserID, report.Fingerprint[:]).Scan(&reportID)
if err == nil {
existing, found, err := getModerationReport(ctx, tx, reportID)
if err != nil {
return domain.ModerationReport{}, false, err
}
if !found {
return domain.ModerationReport{}, false, fmt.Errorf("duplicate moderation report disappeared")
}
return existing, false, nil
}
if !errors.Is(err, pgx.ErrNoRows) {
return domain.ModerationReport{}, false, fmt.Errorf("lookup moderation report fingerprint: %w", err)
}
var hourly, daily int
if err := tx.QueryRow(ctx, `
SELECT
count(*) FILTER (WHERE created_at >= $2::timestamptz - interval '1 hour'),
count(*) FILTER (WHERE created_at >= $2::timestamptz - interval '24 hours')
FROM moderation_reports
WHERE reporter_user_id = $1
AND created_at <= $2::timestamptz`,
report.ReporterUserID, report.CreatedAt).Scan(&hourly, &daily); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("count moderation reporter submissions: %w", err)
}
if hourly >= domain.MaxModerationReportsPerHour || daily >= domain.MaxModerationReportsPerDay {
return domain.ModerationReport{}, false, domain.ErrModerationRateLimited
}
reportID, created, err := insertModerationReport(ctx, tx, report)
if err != nil {
return domain.ModerationReport{}, false, err
}
if !created {
existing, found, err := getModerationReport(ctx, tx, reportID)
if err != nil {
return domain.ModerationReport{}, false, err
}
if !found {
return domain.ModerationReport{}, false, fmt.Errorf("duplicate moderation report disappeared")
}
return existing, false, nil
}
report.ID = reportID
return domain.CloneModerationReport(report), true, nil
}
func (s *ModerationReportStore) ImportLegacyEphemeralReport(ctx context.Context, legacyReportID int64, report domain.ModerationReport) (domain.ModerationReport, bool, error) {
if s == nil || s.db == nil {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store is not configured")
}
if legacyReportID <= 0 {
return domain.ModerationReport{}, false, domain.ErrModerationReportInvalid
}
if err := report.Validate(); err != nil {
return domain.ModerationReport{}, false, err
}
if report.Source != domain.ModerationSourceEphemeral {
return domain.ModerationReport{}, false, domain.ErrModerationReportInvalid
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store requires transaction-capable postgres handle")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("begin legacy ephemeral report import: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
if _, err := tx.Exec(ctx, `
SELECT pg_advisory_xact_lock(
hashtextextended('moderation-legacy-ephemeral:' || $1::bigint::text, 0)
)`, legacyReportID); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("lock legacy ephemeral report: %w", err)
}
var reportID int64
err = tx.QueryRow(ctx, `
SELECT moderation_report_id
FROM moderation_legacy_ephemeral_migrations
WHERE legacy_report_id = $1`, legacyReportID).Scan(&reportID)
if err == nil {
existing, found, err := getModerationReport(ctx, tx, reportID)
if err != nil {
return domain.ModerationReport{}, false, err
}
if !found {
return domain.ModerationReport{}, false, fmt.Errorf("legacy ephemeral report mapping points to missing moderation report")
}
return existing, false, nil
}
if !errors.Is(err, pgx.ErrNoRows) {
return domain.ModerationReport{}, false, fmt.Errorf("lookup legacy ephemeral report mapping: %w", err)
}
if _, err := tx.Exec(ctx, `
SELECT pg_advisory_xact_lock(
hashtextextended('moderation-report:' || $1::bigint::text, 0)
)`, report.ReporterUserID); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("lock moderation reporter: %w", err)
}
reportID, created, err := insertModerationReport(ctx, tx, report)
if err != nil {
return domain.ModerationReport{}, false, err
}
if _, err := tx.Exec(ctx, `
INSERT INTO moderation_legacy_ephemeral_migrations (
legacy_report_id, moderation_report_id, migrated_at
) VALUES ($1,$2,clock_timestamp())`,
legacyReportID, reportID,
); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("insert legacy ephemeral report mapping: %w", err)
}
if err := tx.Commit(ctx); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("commit legacy ephemeral report import: %w", err)
}
if created {
report.ID = reportID
return domain.CloneModerationReport(report), true, nil
}
existing, found, err := getModerationReport(ctx, s.db, reportID)
if err != nil {
return domain.ModerationReport{}, false, err
}
if !found {
return domain.ModerationReport{}, false, fmt.Errorf("imported duplicate moderation report disappeared")
}
return existing, false, nil
}
func insertModerationReport(ctx context.Context, tx pgx.Tx, report domain.ModerationReport) (int64, bool, error) {
var reportID int64
err := tx.QueryRow(ctx, `
INSERT INTO moderation_reports (
reporter_user_id, source, target_peer_type, target_peer_id, reason,
report_option, report_comment, comment_hash, fingerprint,
taxonomy_version, created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
ON CONFLICT (reporter_user_id, fingerprint) DO NOTHING
RETURNING id`,
report.ReporterUserID, string(report.Source), string(report.Target.Type),
report.Target.ID, string(report.Reason), report.Option, report.Comment,
report.CommentHash[:], report.Fingerprint[:], report.TaxonomyVersion,
report.CreatedAt,
).Scan(&reportID)
if errors.Is(err, pgx.ErrNoRows) {
if err := tx.QueryRow(ctx, `
SELECT id
FROM moderation_reports
WHERE reporter_user_id = $1 AND fingerprint = $2`,
report.ReporterUserID, report.Fingerprint[:]).Scan(&reportID); err != nil {
return 0, false, fmt.Errorf("lookup duplicate moderation report: %w", err)
}
return reportID, false, nil
}
if err != nil {
return 0, false, fmt.Errorf("insert moderation report: %w", err)
}
for ordinal, item := range report.Items {
if _, err := tx.Exec(ctx, `
INSERT INTO moderation_report_items (
report_id, ordinal, item_kind, peer_type, peer_id, item_id,
secondary_id, author_user_id, evidence_schema_version, evidence,
evidence_hash
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,$11)`,
reportID, ordinal, string(item.Kind), string(item.Peer.Type),
item.Peer.ID, item.ItemID, item.SecondaryID, item.AuthorUserID,
item.EvidenceSchemaVersion, []byte(item.Evidence), item.EvidenceHash[:],
); err != nil {
return 0, false, fmt.Errorf("insert moderation report item %d: %w", ordinal, err)
}
}
for _, hold := range report.MediaHolds {
if _, err := tx.Exec(ctx, `
INSERT INTO moderation_media_holds (
report_id, item_ordinal, media_kind, storage_key, created_at
) VALUES ($1,$2,$3,$4,$5)`,
reportID, hold.ItemIndex, string(hold.Kind), hold.StorageKey,
report.CreatedAt,
); err != nil {
return 0, false, fmt.Errorf("insert moderation media hold: %w", err)
}
}
if err := attachModerationReportToCase(ctx, tx, reportID, report); err != nil {
return 0, false, err
}
return reportID, true, nil
}
func attachModerationReportToCase(ctx context.Context, tx pgx.Tx, reportID int64, report domain.ModerationReport) error {
if _, err := tx.Exec(ctx, `
SELECT pg_advisory_xact_lock(
hashtextextended(
'moderation-case:' || $1::text || ':' || $2::bigint::text,
0
)
)`, string(report.Target.Type), report.Target.ID); err != nil {
return fmt.Errorf("lock moderation case target: %w", err)
}
var caseID int64
err := tx.QueryRow(ctx, `
SELECT id
FROM moderation_cases
WHERE target_peer_type = $1
AND target_peer_id = $2
AND status IN ('open', 'in_review')
FOR UPDATE`,
string(report.Target.Type), report.Target.ID,
).Scan(&caseID)
if errors.Is(err, pgx.ErrNoRows) {
err = tx.QueryRow(ctx, `
INSERT INTO moderation_cases (
target_peer_type, target_peer_id, status, severity, assigned_to,
version, report_count, distinct_reporter_count, first_report_at,
last_report_at, created_at, updated_at
) VALUES ($1,$2,'open',$3,'',1,1,1,$4,$4,$4,$4)
RETURNING id`,
string(report.Target.Type), report.Target.ID,
int16(domain.ModerationSeverityForReason(report.Reason)),
report.CreatedAt,
).Scan(&caseID)
if err != nil {
return fmt.Errorf("create moderation case: %w", err)
}
if _, err := tx.Exec(ctx, `
INSERT INTO moderation_case_reports (case_id, report_id, attached_at)
VALUES ($1,$2,$3)`, caseID, reportID, report.CreatedAt); err != nil {
return fmt.Errorf("attach report to new moderation case: %w", err)
}
return nil
}
if err != nil {
return fmt.Errorf("find active moderation case: %w", err)
}
if _, err := tx.Exec(ctx, `
INSERT INTO moderation_case_reports (case_id, report_id, attached_at)
VALUES ($1,$2,$3)`, caseID, reportID, report.CreatedAt); err != nil {
return fmt.Errorf("attach report to moderation case: %w", err)
}
if _, err := tx.Exec(ctx, `
UPDATE moderation_cases c
SET severity = greatest(c.severity, $2),
version = c.version + 1,
report_count = (
SELECT count(*)::integer
FROM moderation_case_reports cr
WHERE cr.case_id = c.id
),
distinct_reporter_count = (
SELECT count(DISTINCT r.reporter_user_id)::integer
FROM moderation_case_reports cr
JOIN moderation_reports r ON r.id = cr.report_id
WHERE cr.case_id = c.id
),
first_report_at = least(c.first_report_at, $3),
last_report_at = greatest(c.last_report_at, $3),
updated_at = greatest(c.updated_at, $3)
WHERE c.id = $1`,
caseID, int16(domain.ModerationSeverityForReason(report.Reason)),
report.CreatedAt,
); err != nil {
return fmt.Errorf("update moderation case aggregates: %w", err)
}
return nil
}
func (s *ModerationReportStore) GetModerationReport(ctx context.Context, reportID int64) (domain.ModerationReport, bool, error) {
if s == nil || s.db == nil {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store is not configured")
}
if reportID <= 0 {
return domain.ModerationReport{}, false, domain.ErrModerationReportInvalid
}
return getModerationReport(ctx, s.db, reportID)
}
func getModerationReport(ctx context.Context, db sqlcgen.DBTX, reportID int64) (domain.ModerationReport, bool, error) {
var (
report domain.ModerationReport
source, target, reason string
commentHash []byte
fingerprint []byte
)
err := db.QueryRow(ctx, `
SELECT id, reporter_user_id, source, target_peer_type, target_peer_id,
reason, report_option, report_comment, comment_hash, fingerprint,
taxonomy_version, created_at
FROM moderation_reports
WHERE id = $1`, reportID).Scan(
&report.ID, &report.ReporterUserID, &source, &target,
&report.Target.ID, &reason, &report.Option, &report.Comment,
&commentHash, &fingerprint, &report.TaxonomyVersion,
&report.CreatedAt,
)
if errors.Is(err, pgx.ErrNoRows) {
return domain.ModerationReport{}, false, nil
}
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("get moderation report: %w", err)
}
if len(commentHash) != len(report.CommentHash) || len(fingerprint) != len(report.Fingerprint) {
return domain.ModerationReport{}, false, fmt.Errorf("get moderation report: invalid persisted hash length")
}
copy(report.CommentHash[:], commentHash)
copy(report.Fingerprint[:], fingerprint)
report.Source = domain.ModerationReportSource(source)
report.Target.Type = domain.PeerType(target)
report.Reason = domain.ModerationReason(reason)
rows, err := db.Query(ctx, `
SELECT ordinal, item_kind, peer_type, peer_id, item_id, secondary_id,
author_user_id, evidence_schema_version, evidence, evidence_hash
FROM moderation_report_items
WHERE report_id = $1
ORDER BY ordinal`, reportID)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("list moderation report items: %w", err)
}
for rows.Next() {
var (
ordinal int
item domain.ModerationReportItem
kind, peerType string
evidence, evidenceHash []byte
)
if err := rows.Scan(
&ordinal, &kind, &peerType, &item.Peer.ID, &item.ItemID,
&item.SecondaryID, &item.AuthorUserID,
&item.EvidenceSchemaVersion, &evidence, &evidenceHash,
); err != nil {
rows.Close()
return domain.ModerationReport{}, false, fmt.Errorf("scan moderation report item: %w", err)
}
if ordinal != len(report.Items) || len(evidenceHash) != len(item.EvidenceHash) {
rows.Close()
return domain.ModerationReport{}, false, fmt.Errorf("scan moderation report item: invalid persisted ordering or hash")
}
canonical, err := domain.CanonicalModerationEvidence(evidence)
if err != nil {
rows.Close()
return domain.ModerationReport{}, false, fmt.Errorf("scan moderation report item evidence: %w", err)
}
item.Kind = domain.ModerationReportItemKind(kind)
item.Peer.Type = domain.PeerType(peerType)
item.Evidence = canonical
copy(item.EvidenceHash[:], evidenceHash)
report.Items = append(report.Items, item)
}
if err := rows.Err(); err != nil {
rows.Close()
return domain.ModerationReport{}, false, fmt.Errorf("iterate moderation report items: %w", err)
}
rows.Close()
holdRows, err := db.Query(ctx, `
SELECT item_ordinal, media_kind, storage_key
FROM moderation_media_holds
WHERE report_id = $1
ORDER BY item_ordinal, media_kind, storage_key`, reportID)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("list moderation media holds: %w", err)
}
defer holdRows.Close()
for holdRows.Next() {
var hold domain.ModerationMediaHold
var kind string
if err := holdRows.Scan(&hold.ItemIndex, &kind, &hold.StorageKey); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("scan moderation media hold: %w", err)
}
hold.Kind = domain.ModerationMediaKind(kind)
report.MediaHolds = append(report.MediaHolds, hold)
}
if err := holdRows.Err(); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("iterate moderation media holds: %w", err)
}
if err := report.Validate(); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("validate persisted moderation report: %w", err)
}
return report, true, nil
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,471 @@
package postgres
import (
"context"
"errors"
"sync"
"testing"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
moderationapp "telesrv/internal/app/moderation"
"telesrv/internal/domain"
)
func TestModerationReportStoreAtomicEvidenceAndIdempotency(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
now := time.Now().UTC()
reporter := now.UnixNano()&0x3fffffff + 5_000
report, err := domain.NewModerationReport(domain.ModerationReportDraft{
ReporterUserID: reporter, Source: domain.ModerationSourceProfilePhoto,
Target: domain.Peer{Type: domain.PeerTypeUser, ID: reporter + 1},
Reason: domain.ModerationReasonFake, Option: "v1/fake",
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemProfilePhoto,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: reporter + 1},
ItemID: reporter + 2, AuthorUserID: reporter + 1,
EvidenceSchemaVersion: 1,
Evidence: []byte(`{"photo_id":2,"owner_id":1}`),
}},
MediaHolds: []domain.ModerationMediaHold{{
ItemIndex: 0, Kind: domain.ModerationMediaPhoto,
StorageKey: "profile/photo/test",
}},
CreatedAt: now,
})
if err != nil {
t.Fatal(err)
}
store := NewModerationReportStore(pool)
stored, created, err := store.CreateModerationReport(ctx, report)
if err != nil || !created {
t.Fatalf("create=%v err=%v", created, err)
}
t.Cleanup(func() {
cleanupModerationReport(t, pool, stored.ID)
})
retry, created, err := store.CreateModerationReport(ctx, report)
if err != nil || created || retry.ID != stored.ID {
t.Fatalf("retry=%+v created=%v err=%v", retry, created, err)
}
got, found, err := store.GetModerationReport(ctx, stored.ID)
if err != nil || !found {
t.Fatalf("get found=%v err=%v", found, err)
}
if got.Fingerprint != report.Fingerprint || len(got.Items) != 1 ||
len(got.MediaHolds) != 1 || got.MediaHolds[0].StorageKey != "profile/photo/test" {
t.Fatalf("stored report = %+v", got)
}
var reports, items, holds int
if err := pool.QueryRow(ctx, `
SELECT
(SELECT count(*) FROM moderation_reports WHERE id = $1),
(SELECT count(*) FROM moderation_report_items WHERE report_id = $1),
(SELECT count(*) FROM moderation_media_holds WHERE report_id = $1)`,
stored.ID).Scan(&reports, &items, &holds); err != nil {
t.Fatal(err)
}
if reports != 1 || items != 1 || holds != 1 {
t.Fatalf("rows reports=%d items=%d holds=%d", reports, items, holds)
}
}
func TestModerationSponsoredReportIsAtomicUnderConcurrentFinalOptions(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
now := time.Now().UTC()
userID := now.UnixNano()&0x3fffffff + 8_000
randomID := []byte("postgres-sponsored-random-id")
store := NewModerationReportStore(pool)
impression, err := domain.NewSponsoredMessageImpression(
userID, randomID,
domain.Peer{Type: domain.PeerTypeChannel, ID: userID + 1},
userID+2, []byte(`{"creative_id":"pg-creative","schema_version":1}`),
now, now.Add(time.Hour),
)
if err != nil {
t.Fatal(err)
}
impression, created, err := store.CreateSponsoredMessageImpression(ctx, impression)
if err != nil || !created {
t.Fatalf("impression=%+v created=%v err=%v", impression, created, err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM sponsored_message_impressions WHERE id = $1", impression.ID)
})
service := moderationapp.NewService(store)
type result struct {
report domain.ModerationReport
created bool
err error
}
start := make(chan struct{})
results := make(chan result, 2)
var wg sync.WaitGroup
for _, option := range []struct {
reason domain.ModerationReason
option string
}{
{domain.ModerationReasonSpam, "spam"},
{domain.ModerationReasonFake, "fake"},
} {
wg.Add(1)
go func(reason domain.ModerationReason, option string) {
defer wg.Done()
<-start
report, created, err := service.ReportSponsored(
ctx, userID, randomID, reason, option, now.Add(time.Second),
)
results <- result{report: report, created: created, err: err}
}(option.reason, option.option)
}
close(start)
wg.Wait()
close(results)
var reportID int64
var createdCount int
for got := range results {
if got.err != nil || got.report.ID <= 0 {
t.Fatalf("concurrent result=%+v", got)
}
if reportID == 0 {
reportID = got.report.ID
} else if got.report.ID != reportID {
t.Fatalf("report ids differ: %d vs %d", reportID, got.report.ID)
}
if got.created {
createdCount++
}
}
if createdCount != 1 {
t.Fatalf("created count=%d, want 1", createdCount)
}
t.Cleanup(func() { cleanupModerationReport(t, pool, reportID) })
var reportCount int
if err := pool.QueryRow(ctx, `
SELECT count(*)
FROM moderation_reports
WHERE reporter_user_id = $1 AND source = 'sponsored'`,
userID,
).Scan(&reportCount); err != nil {
t.Fatal(err)
}
if reportCount != 1 {
t.Fatalf("sponsored reports=%d, want 1", reportCount)
}
}
func TestModerationCaseActionAppealLinkAndTelemetryPostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
now := time.Now().UTC()
reporter := now.UnixNano()&0x3fffffff + 12_000
target := domain.Peer{Type: domain.PeerTypeUser, ID: reporter + 1}
store := NewModerationReportStore(pool)
service := moderationapp.NewService(store)
report, err := domain.NewModerationReport(domain.ModerationReportDraft{
ReporterUserID: reporter, Source: domain.ModerationSourceAccountPeer,
Target: target, Reason: domain.ModerationReasonFake, Option: "fake",
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemPeer, Peer: target, ItemID: target.ID,
AuthorUserID: target.ID, EvidenceSchemaVersion: 1,
Evidence: []byte(`{"schema_version":1}`),
}},
CreatedAt: now,
})
if err != nil {
t.Fatal(err)
}
stored, created, err := store.CreateModerationReport(ctx, report)
if err != nil || !created {
t.Fatalf("report=%+v created=%v err=%v", stored, created, err)
}
t.Cleanup(func() { cleanupModerationReport(t, pool, stored.ID) })
cases, err := store.ListModerationCases(ctx, domain.ModerationCaseFilter{
Target: target, Limit: 10,
})
if err != nil || len(cases) != 1 {
t.Fatalf("cases=%+v err=%v", cases, err)
}
claimed, err := store.ClaimModerationCase(
ctx, cases[0].ID, cases[0].Version, "pg-reviewer", now.Add(time.Second),
)
if err != nil {
t.Fatal(err)
}
decision, err := domain.NewModerationDecisionRequest(domain.ModerationDecisionRequest{
CaseID: claimed.ID, ExpectedVersion: claimed.Version,
Actor: "pg-reviewer", Reason: "confirmed fake",
CommandID: "pg-moderation-decision-" + time.Unix(0, reporter).Format("150405.000000000"),
Kind: domain.ModerationDecisionViolation,
Actions: []domain.ModerationActionDraft{{
Kind: domain.ModerationActionMarkFake, Payload: []byte(`{}`),
}},
CreatedAt: now.Add(2 * time.Second),
})
if err != nil {
t.Fatal(err)
}
if _, created, err := store.DecideModerationCase(ctx, decision); err != nil || !created {
t.Fatalf("decision created=%v err=%v", created, err)
}
actions, err := store.ClaimModerationActions(
ctx, now.Add(3*time.Second), 10, time.Minute,
)
if err != nil || len(actions) != 1 {
t.Fatalf("actions=%+v err=%v", actions, err)
}
if err := store.CompleteModerationAction(
ctx, actions[0].ID, actions[0].Attempts, true, "",
time.Time{}, now.Add(4*time.Second),
); err != nil {
t.Fatal(err)
}
token, err := service.IssueAppealLink(
ctx, cases[0].ID, target.ID, now.Add(time.Hour), now.Add(5*time.Second),
)
if err != nil {
t.Fatal(err)
}
appeal, created, err := service.SubmitAppealLink(
ctx, token, "Postgres appeal.", now.Add(6*time.Second),
)
if err != nil || !created || appeal.ID <= 0 {
t.Fatalf("appeal=%+v created=%v err=%v", appeal, created, err)
}
retry, created, err := service.SubmitAppealLink(
ctx, token, "retry body", now.Add(7*time.Second),
)
if err != nil || created || retry.ID != appeal.ID ||
retry.Text != appeal.Text {
t.Fatalf("appeal retry=%+v created=%v err=%v", retry, created, err)
}
telemetryStore := NewClientTelemetryStore(pool)
telemetryAt := time.Unix(reporter%1_000_000+1, 0).UTC()
event, err := domain.NewClientTelemetryEvent(
reporter, domain.ClientTelemetryMessageDelivery, target,
[]int64{3, 1, 2}, map[string]any{"push": true}, telemetryAt,
)
if err != nil {
t.Fatal(err)
}
telemetry, created, err := telemetryStore.CreateClientTelemetry(ctx, event)
if err != nil || !created || telemetry.ID <= 0 {
t.Fatalf("telemetry=%+v created=%v err=%v", telemetry, created, err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM client_telemetry_events WHERE id = $1", telemetry.ID)
})
retryTelemetry, created, err := telemetryStore.CreateClientTelemetry(ctx, event)
if err != nil || created || retryTelemetry.ID != telemetry.ID {
t.Fatalf("telemetry retry=%+v created=%v err=%v", retryTelemetry, created, err)
}
deleted, err := telemetryStore.DeleteExpiredClientTelemetry(
ctx, telemetryAt.Add(time.Second), 10,
)
if err != nil || deleted < 1 {
t.Fatalf("telemetry retention deleted=%d err=%v", deleted, err)
}
}
func TestModerationSanctionSupersessionAndAppealOwnershipPostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
now := time.Now().UTC()
store := NewModerationReportStore(pool)
service := moderationapp.NewService(store)
base := now.UnixNano()&0x3fffffff + 40_000
createDecision := func(target domain.Peer, reporter int64, option, command string, at time.Time) (int64, int64) {
t.Helper()
report, _, err := service.AcceptReport(ctx, domain.ModerationReportDraft{
ReporterUserID: reporter, Source: domain.ModerationSourceAccountPeer,
Target: target, Reason: domain.ModerationReasonFake, Option: option,
Items: []domain.ModerationReportItem{{
Kind: domain.ModerationItemPeer, Peer: target, ItemID: target.ID,
AuthorUserID: target.ID, EvidenceSchemaVersion: 1,
Evidence: []byte(`{"schema_version":1}`),
}},
CreatedAt: at,
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { cleanupModerationReport(t, pool, report.ID) })
cases, err := service.ListCases(ctx, domain.ModerationCaseFilter{
Statuses: []domain.ModerationCaseStatus{domain.ModerationCaseOpen},
Target: target, Limit: 10,
})
if err != nil || len(cases) != 1 {
t.Fatalf("open cases=%+v err=%v", cases, err)
}
claimed, err := service.ClaimCase(
ctx, cases[0].ID, cases[0].Version, "pg-owner", at.Add(time.Second),
)
if err != nil {
t.Fatal(err)
}
detail, _, err := service.DecideCase(ctx, domain.ModerationDecisionRequest{
CaseID: claimed.ID, ExpectedVersion: claimed.Version,
Actor: "pg-owner", Reason: "confirmed", CommandID: command,
Kind: domain.ModerationDecisionViolation,
Actions: []domain.ModerationActionDraft{{
Kind: domain.ModerationActionMarkFake, Payload: []byte(`{}`),
}},
CreatedAt: at.Add(2 * time.Second),
})
if err != nil || len(detail.Actions) != 1 {
t.Fatalf("decision=%+v err=%v", detail, err)
}
return claimed.ID, detail.Actions[0].ID
}
target := domain.Peer{Type: domain.PeerTypeUser, ID: base + 1}
oldCaseID, oldActionID := createDecision(target, base+2, "old", "pg-old", now)
newCaseID, newActionID := createDecision(target, base+3, "new", "pg-new", now.Add(3*time.Second))
claimedActions, err := store.ClaimModerationActions(ctx, now.Add(6*time.Second), 10, time.Minute)
if err != nil {
t.Fatal(err)
}
claimedByID := make(map[int64]domain.ModerationAction, len(claimedActions))
for _, action := range claimedActions {
claimedByID[action.ID] = action
}
oldAction, oldFound := claimedByID[oldActionID]
newAction, newFound := claimedByID[newActionID]
if !oldFound || !newFound {
t.Fatalf("claimed actions=%+v", claimedActions)
}
if current, err := store.IsModerationActionCurrent(ctx, oldAction); err != nil || current {
t.Fatalf("old current=%v err=%v", current, err)
}
if current, err := store.IsModerationActionCurrent(ctx, newAction); err != nil || !current {
t.Fatalf("new current=%v err=%v", current, err)
}
if err := store.SupersedeModerationAction(
ctx, oldAction.ID, oldAction.Attempts, now.Add(7*time.Second),
); err != nil {
t.Fatal(err)
}
if err := store.CompleteModerationAction(
ctx, newAction.ID, newAction.Attempts, true, "", time.Time{},
now.Add(8*time.Second),
); err != nil {
t.Fatal(err)
}
oldDetail, _, err := service.Case(ctx, oldCaseID)
if err != nil || oldDetail.Case.Status != domain.ModerationCaseResolved ||
oldDetail.Actions[0].Status != domain.ModerationActionSuperseded {
t.Fatalf("old detail=%+v err=%v", oldDetail, err)
}
newDetail, _, err := service.Case(ctx, newCaseID)
if err != nil || newDetail.Case.Status != domain.ModerationCaseResolved ||
newDetail.Actions[0].Status != domain.ModerationActionSucceeded {
t.Fatalf("new detail=%+v err=%v", newDetail, err)
}
appealTarget := domain.Peer{Type: domain.PeerTypeUser, ID: base + 10}
appealedCaseID, appealedActionID := createDecision(
appealTarget, base+11, "appealed", "pg-appealed", now.Add(10*time.Second),
)
actions, err := store.ClaimModerationActions(ctx, now.Add(13*time.Second), 10, time.Minute)
if err != nil || len(actions) != 1 || actions[0].ID != appealedActionID {
t.Fatalf("appealed action=%+v err=%v", actions, err)
}
if err := store.CompleteModerationAction(
ctx, actions[0].ID, actions[0].Attempts, true, "", time.Time{},
now.Add(14*time.Second),
); err != nil {
t.Fatal(err)
}
appeal, _, err := service.SubmitAppeal(
ctx, appealedCaseID, appealTarget.ID, "please review", now.Add(15*time.Second),
)
if err != nil {
t.Fatal(err)
}
_, _ = createDecision(
appealTarget, base+12, "newer", "pg-newer-owner", now.Add(16*time.Second),
)
appealedDetail, _, err := service.Case(ctx, appealedCaseID)
if err != nil {
t.Fatal(err)
}
appealClaim, err := service.ClaimCase(
ctx, appealedCaseID, appealedDetail.Case.Version, "pg-owner", now.Add(19*time.Second),
)
if err != nil {
t.Fatal(err)
}
_, _, err = service.ReviewAppeal(ctx, domain.ModerationDecisionRequest{
CaseID: appealedCaseID, AppealID: appeal.ID,
ExpectedVersion: appealClaim.Version, Actor: "pg-owner",
Reason: "grant", CommandID: "pg-stale-appeal",
Kind: domain.ModerationDecisionAppealGrant,
Actions: []domain.ModerationActionDraft{{
Kind: domain.ModerationActionClearPeerFlags, Payload: []byte(`{}`),
}},
CreatedAt: now.Add(20 * time.Second),
})
if !errors.Is(err, domain.ErrModerationActionConflict) {
t.Fatalf("ReviewAppeal error=%v", err)
}
}
func cleanupModerationReport(t *testing.T, pool *pgxpool.Pool, reportID int64) {
t.Helper()
ctx := context.Background()
tx, err := pool.Begin(ctx)
if err != nil {
t.Errorf("begin moderation cleanup: %v", err)
return
}
defer func() { _ = tx.Rollback(ctx) }()
var caseID int64
err = tx.QueryRow(ctx, `
SELECT case_id FROM moderation_case_reports WHERE report_id = $1`,
reportID,
).Scan(&caseID)
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
t.Errorf("find moderation cleanup case: %v", err)
return
}
if _, err := tx.Exec(ctx, "DELETE FROM sponsored_message_impressions WHERE report_id = $1", reportID); err != nil {
t.Errorf("cleanup sponsored impression: %v", err)
return
}
if _, err := tx.Exec(ctx, "DELETE FROM channel_antispam_decisions WHERE report_id = $1", reportID); err != nil {
t.Errorf("cleanup anti-spam decision: %v", err)
return
}
if caseID > 0 {
for _, statement := range []string{
"DELETE FROM moderation_actions WHERE case_id = $1",
"DELETE FROM moderation_decisions WHERE case_id = $1",
"DELETE FROM moderation_appeal_links WHERE case_id = $1",
"DELETE FROM moderation_appeals WHERE case_id = $1",
"DELETE FROM moderation_case_reports WHERE case_id = $1",
"DELETE FROM moderation_cases WHERE id = $1",
} {
if _, err := tx.Exec(ctx, statement, caseID); err != nil {
t.Errorf("moderation cleanup %q: %v", statement, err)
return
}
}
}
if _, err := tx.Exec(ctx, "DELETE FROM moderation_legacy_ephemeral_migrations WHERE moderation_report_id = $1", reportID); err != nil {
t.Errorf("cleanup legacy moderation mapping: %v", err)
return
}
if _, err := tx.Exec(ctx, "DELETE FROM moderation_reports WHERE id = $1", reportID); err != nil {
t.Errorf("cleanup moderation report: %v", err)
return
}
if err := tx.Commit(ctx); err != nil {
t.Errorf("commit moderation cleanup: %v", err)
}
}

View file

@ -0,0 +1,353 @@
package postgres
import (
"context"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"telesrv/internal/domain"
)
func (s *ModerationReportStore) CreateSponsoredMessageImpression(ctx context.Context, impression domain.SponsoredMessageImpression) (domain.SponsoredMessageImpression, bool, error) {
if s == nil || s.db == nil {
return domain.SponsoredMessageImpression{}, false, fmt.Errorf("moderation report store is not configured")
}
if err := impression.Validate(); err != nil || impression.ID != 0 ||
impression.ReportID != 0 {
return domain.SponsoredMessageImpression{}, false, domain.ErrModerationReportInvalid
}
err := s.db.QueryRow(ctx, `
INSERT INTO sponsored_message_impressions (
user_id, random_id_hash, target_peer_type, target_peer_id,
author_user_id, evidence_schema_version, evidence, evidence_hash,
created_at, expires_at
) VALUES ($1,$2,$3,$4,$5,$6,$7::jsonb,$8,$9,$10)
ON CONFLICT (user_id, random_id_hash) DO NOTHING
RETURNING id`,
impression.UserID, impression.RandomIDHash[:],
string(impression.Target.Type), impression.Target.ID,
impression.AuthorUserID, impression.EvidenceSchemaVersion,
[]byte(impression.Evidence), impression.EvidenceHash[:],
impression.CreatedAt, impression.ExpiresAt,
).Scan(&impression.ID)
if err == nil {
return impression, true, nil
}
if !errors.Is(err, pgx.ErrNoRows) {
return domain.SponsoredMessageImpression{}, false, fmt.Errorf("insert sponsored impression: %w", err)
}
existing, found, err := s.GetSponsoredMessageImpression(
ctx, impression.UserID, impression.RandomIDHash, impression.CreatedAt,
)
if err != nil {
return domain.SponsoredMessageImpression{}, false, err
}
if !found || existing.Target != impression.Target ||
existing.AuthorUserID != impression.AuthorUserID ||
existing.EvidenceHash != impression.EvidenceHash ||
!existing.ExpiresAt.Equal(impression.ExpiresAt) {
return domain.SponsoredMessageImpression{}, false, domain.ErrModerationActionConflict
}
return existing, false, nil
}
func (s *ModerationReportStore) GetSponsoredMessageImpression(ctx context.Context, userID int64, randomIDHash [32]byte, now time.Time) (domain.SponsoredMessageImpression, bool, error) {
if s == nil || s.db == nil {
return domain.SponsoredMessageImpression{}, false, fmt.Errorf("moderation report store is not configured")
}
if userID <= 0 || randomIDHash == ([32]byte{}) || now.IsZero() {
return domain.SponsoredMessageImpression{}, false, domain.ErrModerationReportInvalid
}
impression, err := scanSponsoredMessageImpression(s.db.QueryRow(ctx, `
SELECT id, user_id, random_id_hash, target_peer_type, target_peer_id,
author_user_id, evidence_schema_version, evidence, evidence_hash,
report_id, created_at, expires_at
FROM sponsored_message_impressions
WHERE user_id = $1 AND random_id_hash = $2 AND expires_at > $3`,
userID, randomIDHash[:], now,
))
if errors.Is(err, pgx.ErrNoRows) {
return domain.SponsoredMessageImpression{}, false, nil
}
if err != nil {
return domain.SponsoredMessageImpression{}, false, fmt.Errorf("get sponsored impression: %w", err)
}
return impression, true, nil
}
func (s *ModerationReportStore) CreateSponsoredModerationReport(ctx context.Context, impressionID int64, report domain.ModerationReport) (domain.ModerationReport, bool, error) {
if s == nil || s.db == nil {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store is not configured")
}
if impressionID <= 0 {
return domain.ModerationReport{}, false, domain.ErrModerationReportInvalid
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store requires transaction-capable postgres handle")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("begin sponsored moderation report: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
impression, err := scanSponsoredMessageImpression(tx.QueryRow(ctx, `
SELECT id, user_id, random_id_hash, target_peer_type, target_peer_id,
author_user_id, evidence_schema_version, evidence, evidence_hash,
report_id, created_at, expires_at
FROM sponsored_message_impressions
WHERE id = $1
FOR UPDATE`, impressionID))
if errors.Is(err, pgx.ErrNoRows) {
return domain.ModerationReport{}, false, domain.ErrModerationEvidenceNotFound
}
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("lock sponsored impression: %w", err)
}
if !report.CreatedAt.Before(impression.ExpiresAt) {
return domain.ModerationReport{}, false, domain.ErrModerationImpressionExpired
}
if err := domain.ValidateSponsoredModerationReport(impression, report); err != nil {
return domain.ModerationReport{}, false, err
}
if impression.ReportID > 0 {
existing, found, err := getModerationReport(ctx, tx, impression.ReportID)
if err != nil {
return domain.ModerationReport{}, false, err
}
if !found {
return domain.ModerationReport{}, false, domain.ErrModerationReportNotFound
}
return existing, false, nil
}
stored, created, err := createModerationReportTx(ctx, tx, report)
if err != nil {
return domain.ModerationReport{}, false, err
}
tag, err := tx.Exec(ctx, `
UPDATE sponsored_message_impressions
SET report_id = $2
WHERE id = $1 AND report_id IS NULL`, impressionID, stored.ID)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("link sponsored report: %w", err)
}
if tag.RowsAffected() != 1 {
return domain.ModerationReport{}, false, domain.ErrModerationActionConflict
}
if err := tx.Commit(ctx); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("commit sponsored moderation report: %w", err)
}
return stored, created, nil
}
func (s *ModerationReportStore) CreateChannelAntiSpamDecision(ctx context.Context, decision domain.ChannelAntiSpamDecision) (domain.ChannelAntiSpamDecision, bool, error) {
if s == nil || s.db == nil {
return domain.ChannelAntiSpamDecision{}, false, fmt.Errorf("moderation report store is not configured")
}
if err := decision.Validate(); err != nil || decision.ID != 0 ||
decision.ReportID != 0 {
return domain.ChannelAntiSpamDecision{}, false, domain.ErrModerationReportInvalid
}
err := s.db.QueryRow(ctx, `
INSERT INTO channel_antispam_decisions (
channel_id, message_id, author_user_id, evidence_schema_version,
evidence, evidence_hash, created_at
) VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7)
ON CONFLICT (channel_id, message_id) DO NOTHING
RETURNING id`,
decision.ChannelID, decision.MessageID, decision.AuthorUserID,
decision.EvidenceSchemaVersion, []byte(decision.Evidence),
decision.EvidenceHash[:], decision.CreatedAt,
).Scan(&decision.ID)
if err == nil {
return decision, true, nil
}
if !errors.Is(err, pgx.ErrNoRows) {
return domain.ChannelAntiSpamDecision{}, false, fmt.Errorf("insert anti-spam decision: %w", err)
}
existing, found, err := s.GetChannelAntiSpamDecision(
ctx, decision.ChannelID, decision.MessageID,
)
if err != nil {
return domain.ChannelAntiSpamDecision{}, false, err
}
if !found || existing.AuthorUserID != decision.AuthorUserID ||
existing.EvidenceHash != decision.EvidenceHash {
return domain.ChannelAntiSpamDecision{}, false, domain.ErrModerationActionConflict
}
return existing, false, nil
}
func (s *ModerationReportStore) GetChannelAntiSpamDecision(ctx context.Context, channelID int64, messageID int) (domain.ChannelAntiSpamDecision, bool, error) {
if s == nil || s.db == nil {
return domain.ChannelAntiSpamDecision{}, false, fmt.Errorf("moderation report store is not configured")
}
if channelID <= 0 || messageID <= 0 || messageID > domain.MaxMessageBoxID {
return domain.ChannelAntiSpamDecision{}, false, domain.ErrModerationReportInvalid
}
decision, err := scanChannelAntiSpamDecision(s.db.QueryRow(ctx, `
SELECT id, channel_id, message_id, author_user_id,
evidence_schema_version, evidence, evidence_hash, report_id,
created_at
FROM channel_antispam_decisions
WHERE channel_id = $1 AND message_id = $2`, channelID, messageID))
if errors.Is(err, pgx.ErrNoRows) {
return domain.ChannelAntiSpamDecision{}, false, nil
}
if err != nil {
return domain.ChannelAntiSpamDecision{}, false, fmt.Errorf("get anti-spam decision: %w", err)
}
return decision, true, nil
}
func (s *ModerationReportStore) CreateAntiSpamFalsePositiveReport(ctx context.Context, decisionID int64, report domain.ModerationReport) (domain.ModerationReport, bool, error) {
if s == nil || s.db == nil {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store is not configured")
}
if decisionID <= 0 {
return domain.ModerationReport{}, false, domain.ErrModerationReportInvalid
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.ModerationReport{}, false, fmt.Errorf("moderation report store requires transaction-capable postgres handle")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("begin anti-spam false-positive report: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
decision, err := scanChannelAntiSpamDecision(tx.QueryRow(ctx, `
SELECT id, channel_id, message_id, author_user_id,
evidence_schema_version, evidence, evidence_hash, report_id,
created_at
FROM channel_antispam_decisions
WHERE id = $1
FOR UPDATE`, decisionID))
if errors.Is(err, pgx.ErrNoRows) {
return domain.ModerationReport{}, false, domain.ErrModerationEvidenceNotFound
}
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("lock anti-spam decision: %w", err)
}
if err := domain.ValidateAntiSpamFalsePositiveReport(decision, report); err != nil {
return domain.ModerationReport{}, false, err
}
if decision.ReportID > 0 {
existing, found, err := getModerationReport(ctx, tx, decision.ReportID)
if err != nil {
return domain.ModerationReport{}, false, err
}
if !found {
return domain.ModerationReport{}, false, domain.ErrModerationReportNotFound
}
return existing, false, nil
}
stored, created, err := createModerationReportTx(ctx, tx, report)
if err != nil {
return domain.ModerationReport{}, false, err
}
tag, err := tx.Exec(ctx, `
UPDATE channel_antispam_decisions
SET report_id = $2
WHERE id = $1 AND report_id IS NULL`, decisionID, stored.ID)
if err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("link anti-spam report: %w", err)
}
if tag.RowsAffected() != 1 {
return domain.ModerationReport{}, false, domain.ErrModerationActionConflict
}
if err := tx.Commit(ctx); err != nil {
return domain.ModerationReport{}, false, fmt.Errorf("commit anti-spam false-positive report: %w", err)
}
return stored, created, nil
}
func (s *ModerationReportStore) DeleteExpiredSponsoredMessageImpressions(ctx context.Context, olderThan time.Time, limit int) (int, error) {
if s == nil || s.db == nil {
return 0, fmt.Errorf("moderation report store is not configured")
}
if olderThan.IsZero() || limit <= 0 || limit > 10000 {
return 0, domain.ErrModerationReportInvalid
}
tag, err := s.db.Exec(ctx, `
WITH doomed AS (
SELECT id
FROM sponsored_message_impressions
WHERE expires_at < $1
ORDER BY expires_at, id
LIMIT $2
)
DELETE FROM sponsored_message_impressions i
USING doomed d
WHERE i.id = d.id`, olderThan, limit)
if err != nil {
return 0, fmt.Errorf("delete expired sponsored impressions: %w", err)
}
return int(tag.RowsAffected()), nil
}
func scanSponsoredMessageImpression(row moderationCaseScanner) (domain.SponsoredMessageImpression, error) {
var impression domain.SponsoredMessageImpression
var randomIDHash, evidence, evidenceHash []byte
var peerType string
var reportID *int64
if err := row.Scan(
&impression.ID, &impression.UserID, &randomIDHash, &peerType,
&impression.Target.ID, &impression.AuthorUserID,
&impression.EvidenceSchemaVersion, &evidence, &evidenceHash,
&reportID, &impression.CreatedAt, &impression.ExpiresAt,
); err != nil {
return domain.SponsoredMessageImpression{}, err
}
if len(randomIDHash) != len(impression.RandomIDHash) ||
len(evidenceHash) != len(impression.EvidenceHash) {
return domain.SponsoredMessageImpression{}, domain.ErrModerationReportInvalid
}
copy(impression.RandomIDHash[:], randomIDHash)
copy(impression.EvidenceHash[:], evidenceHash)
impression.Target.Type = domain.PeerType(peerType)
canonical, err := domain.CanonicalModerationEvidence(evidence)
if err != nil {
return domain.SponsoredMessageImpression{}, err
}
impression.Evidence = canonical
if reportID != nil {
impression.ReportID = *reportID
}
if err := impression.Validate(); err != nil {
return domain.SponsoredMessageImpression{}, err
}
return impression, nil
}
func scanChannelAntiSpamDecision(row moderationCaseScanner) (domain.ChannelAntiSpamDecision, error) {
var decision domain.ChannelAntiSpamDecision
var evidence, evidenceHash []byte
var reportID *int64
if err := row.Scan(
&decision.ID, &decision.ChannelID, &decision.MessageID,
&decision.AuthorUserID, &decision.EvidenceSchemaVersion,
&evidence, &evidenceHash, &reportID, &decision.CreatedAt,
); err != nil {
return domain.ChannelAntiSpamDecision{}, err
}
if len(evidenceHash) != len(decision.EvidenceHash) {
return domain.ChannelAntiSpamDecision{}, domain.ErrModerationReportInvalid
}
copy(decision.EvidenceHash[:], evidenceHash)
canonical, err := domain.CanonicalModerationEvidence(evidence)
if err != nil {
return domain.ChannelAntiSpamDecision{}, err
}
decision.Evidence = canonical
if reportID != nil {
decision.ReportID = *reportID
}
if err := decision.Validate(); err != nil {
return domain.ChannelAntiSpamDecision{}, err
}
return decision, nil
}

View file

@ -24,6 +24,16 @@ func NewPrivacyStore(db sqlcgen.DBTX) *PrivacyStore {
return &PrivacyStore{db: db}
}
func (s *PrivacyStore) SupportsDurablePrivacyUpdates() bool {
if s == nil {
return false
}
_, ok := s.db.(interface {
Begin(context.Context) (pgx.Tx, error)
})
return ok
}
func (s *PrivacyStore) GetPrivacyRules(ctx context.Context, ownerUserID int64, key domain.PrivacyKey) (domain.PrivacyRules, bool, error) {
row := s.db.QueryRow(ctx, `
SELECT rules::text
@ -46,11 +56,15 @@ WHERE owner_user_id = $1
}
func (s *PrivacyStore) SetPrivacyRules(ctx context.Context, rules domain.PrivacyRules) error {
return setPrivacyRules(ctx, s.db, rules)
}
func setPrivacyRules(ctx context.Context, db sqlcgen.DBTX, rules domain.PrivacyRules) error {
raw, err := json.Marshal(rules.Rules)
if err != nil {
return err
}
_, err = s.db.Exec(ctx, `
_, err = db.Exec(ctx, `
INSERT INTO account_privacy_rules (owner_user_id, privacy_key, rules, updated_at)
VALUES ($1, $2, $3::jsonb, NOW())
ON CONFLICT (owner_user_id, privacy_key) DO UPDATE SET
@ -63,6 +77,56 @@ ON CONFLICT (owner_user_id, privacy_key) DO UPDATE SET
return nil
}
// SetPrivacyRulesWithUpdate commits the mutable rule row and the immutable
// account update snapshot in one transaction. A privacy rule can therefore
// never become visible without a matching pts event/outbox item.
func (s *PrivacyStore) SetPrivacyRulesWithUpdate(
ctx context.Context,
rules domain.PrivacyRules,
event domain.UpdateEvent,
excludeAuthKeyID [8]byte,
excludeSessionID int64,
) (domain.UpdateEvent, error) {
beginner, ok := s.db.(interface {
Begin(context.Context) (pgx.Tx, error)
})
if !ok {
return domain.UpdateEvent{}, fmt.Errorf("privacy update transaction unavailable")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.UpdateEvent{}, fmt.Errorf("begin privacy update: %w", err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback(ctx)
}
}()
if err := setPrivacyRules(ctx, tx, rules); err != nil {
return domain.UpdateEvent{}, err
}
if event.Date == 0 {
return domain.UpdateEvent{}, fmt.Errorf("privacy update date is required")
}
event.Type = domain.UpdateEventPrivacy
event.Privacy = rules
event.PtsCount = 1
qtx := sqlcgen.New(tx)
recorded, err := NewUpdateEventStore(tx).appendInTx(
ctx, tx, qtx, rules.OwnerUserID, event, true,
excludeAuthKeyID, excludeSessionID, true,
)
if err != nil {
return domain.UpdateEvent{}, fmt.Errorf("append privacy update: %w", err)
}
if err := tx.Commit(ctx); err != nil {
return domain.UpdateEvent{}, fmt.Errorf("commit privacy update: %w", err)
}
committed = true
return recorded, nil
}
func (s *PrivacyStore) ListPrivacyRules(ctx context.Context, ownerUserIDs []int64, keys []domain.PrivacyKey) ([]domain.PrivacyRules, error) {
if len(ownerUserIDs) == 0 || len(keys) == 0 {
return nil, nil

View file

@ -101,6 +101,15 @@ type PrivacyReadModelWarmer interface {
WarmOwners(context.Context, ...int64) error
}
type PrivacyViewerFactsReadModelCache interface {
InvalidateViewerFacts(...int64)
}
type PrivacyMembershipReadModelCache interface {
InvalidateMembership(channelID, userID int64)
InvalidateChannelMemberships(channelID int64)
}
type ProfilePhotoReadModelCache interface {
InvalidateOwner(domain.PeerType, int64)
FlushReadModelCache()
@ -358,6 +367,9 @@ func (l *ReadModelChangeListener) handlePayload(payload string) {
if l.caches.BotProfiles != nil {
l.caches.BotProfiles.InvalidateBotProfileReadModel(evt.PeerID)
}
if cache, ok := l.caches.Privacy.(PrivacyViewerFactsReadModelCache); ok {
cache.InvalidateViewerFacts(evt.PeerID)
}
}
case "user_visibility":
if evt.PeerType == "user" && evt.PeerID != 0 {
@ -457,6 +469,9 @@ func (l *ReadModelChangeListener) handlePayload(payload string) {
if l.caches.RPCProjections != nil {
l.caches.RPCProjections.InvalidateRPCProjectionReadModelForChannel(evt.PeerID)
}
if cache, ok := l.caches.Privacy.(PrivacyMembershipReadModelCache); ok {
cache.InvalidateChannelMemberships(evt.PeerID)
}
}
case "channel_media_counts":
if evt.PeerType == "channel" && evt.PeerID != 0 && l.caches.ChannelMediaCounts != nil {
@ -487,6 +502,9 @@ func (l *ReadModelChangeListener) handlePayload(payload string) {
l.caches.RPCProjections.InvalidateRPCProjectionReadModelForPeer(evt.OwnerUserID, domain.Peer{Type: domain.PeerTypeChannel, ID: evt.PeerID})
l.caches.RPCProjections.InvalidateRPCProjectionReadModelForUser(evt.OwnerUserID)
}
if cache, ok := l.caches.Privacy.(PrivacyMembershipReadModelCache); ok {
cache.InvalidateMembership(evt.PeerID, evt.OwnerUserID)
}
}
case "channel_self_boosts":
if evt.PeerType == "channel" && evt.PeerID != 0 && l.caches.ChannelBoosts != nil {

View file

@ -431,8 +431,8 @@ func (s *StarGiftLifecycleStore) TransferStarGift(ctx context.Context, req domai
return domain.ErrStarGiftTransferUnavailable
}
if _, err := tx.Exec(ctx, `UPDATE peer_star_gifts SET owner_peer_type='user',owner_peer_id=$2,from_user_id=$3,
msg_id=$4,saved_id=0,upgrade_msg_id=$4,gift_date=$5,name_hidden=false,unsaved=false,pinned_order=0,
can_transfer_at=0 WHERE id=$1`, result.Saved.ID, req.To.ID, req.ActorUserID, msgID, req.Date); err != nil {
msg_id=$4,saved_id=0,upgrade_msg_id=$4,gift_date=$5,name_hidden=false,unsaved=$6,pinned_order=0,
can_transfer_at=0 WHERE id=$1`, result.Saved.ID, req.To.ID, req.ActorUserID, msgID, req.Date, req.RecipientUnsaved); err != nil {
return err
}
if err := registerUserStarGiftMessageRef(ctx, tx, req.To.ID, msgID, result.Saved.ID, result.Unique.ID); err != nil {
@ -446,6 +446,7 @@ func (s *StarGiftLifecycleStore) TransferStarGift(ctx context.Context, req domai
}
result.Saved.MsgID, result.Saved.SavedID, result.Saved.UpgradeMsgID, result.Saved.Date = msgID, 0, msgID, req.Date
result.Saved.FromUserID = req.ActorUserID
result.Saved.Unsaved = req.RecipientUnsaved
if sourceSaved.Owner.Type == domain.PeerTypeUser {
_, err := s.retireUserStarGiftMessagesTx(ctx, tx, sourceSaved, result.Unique, req.Date)
return err
@ -587,10 +588,12 @@ func (s *StarGiftLifecycleStore) PurchaseResaleStarGift(ctx context.Context, req
return domain.ErrStarGiftResaleUnavailable
}
if _, err := tx.Exec(ctx, `UPDATE peer_star_gifts SET owner_peer_type=$2,owner_peer_id=$3,from_user_id=$4,
msg_id=$5,saved_id=$6,upgrade_msg_id=$5,gift_date=$7,name_hidden=false,unsaved=false,pinned_order=0,can_transfer_at=0
WHERE id=$1`, result.Saved.ID, string(req.To.Type), req.To.ID, messageSenderID, msgID, savedID, req.Date); err != nil {
msg_id=$5,saved_id=$6,upgrade_msg_id=$5,gift_date=$7,name_hidden=false,unsaved=$8,pinned_order=0,can_transfer_at=0
WHERE id=$1`, result.Saved.ID, string(req.To.Type), req.To.ID, messageSenderID, msgID, savedID, req.Date,
req.To.Type == domain.PeerTypeUser && req.RecipientUnsaved); err != nil {
return err
}
result.Saved.Unsaved = req.To.Type == domain.PeerTypeUser && req.RecipientUnsaved
if req.To.Type == domain.PeerTypeUser {
if err := registerUserStarGiftMessageRef(ctx, tx, req.To.ID, msgID, result.Saved.ID, result.Unique.ID); err != nil {
return err

View file

@ -14,7 +14,7 @@ func TestStarGiftLifecycleMigrationsApply(t *testing.T) {
if err != nil {
t.Fatalf("migrate star gift lifecycle schema: %v", err)
}
if status.Dirty || status.Empty || status.Version != 135 {
t.Fatalf("migration status = %+v, want clean version 135", status)
if status.Dirty || status.Empty || status.Version != 145 {
t.Fatalf("migration status = %+v, want clean version 145", status)
}
}

View file

@ -278,7 +278,7 @@ last_sale_date=$2,updated_at=now() WHERE gift_id=$1`, gift.ID, req.Date); err !=
}
saved := domain.SavedStarGift{Owner: req.To, FromUserID: req.BuyerUserID, GiftID: gift.ID, RevisionID: gift.RevisionID,
Date: req.Date, NameHidden: req.HideName, ConvertStars: gift.ConvertStars, PrepaidUpgradeStars: upgradePrice,
PrepaidUpgradeHash: prepayHash, Message: req.Message}
PrepaidUpgradeHash: prepayHash, Message: req.Message, Unsaved: req.RecipientUnsaved}
return gift, saved, balance, nil
}

View file

@ -152,6 +152,7 @@ WHERE collectible_revision_id=$1 AND crafted
MsgID: ownerMessageID,
Date: req.Date,
NameHidden: req.HideName,
Unsaved: req.RecipientUnsaved,
LifecycleStatus: domain.StarGiftLifecycleActive,
Message: req.Message,
TransferStars: s.lifecycle.TransferStars,

View file

@ -244,12 +244,35 @@ func appendUserUpdateEvent(ctx context.Context, db sqlcgen.DBTX, q *sqlcgen.Quer
}); err != nil {
return err
}
if err := appendPrivacyPayload(ctx, db, userID, event); err != nil {
return err
}
if err := appendQuickReplyPayload(ctx, db, userID, event); err != nil {
return err
}
return nil
}
func appendPrivacyPayload(ctx context.Context, db sqlcgen.DBTX, userID int64, event domain.UpdateEvent) error {
if event.Type != domain.UpdateEventPrivacy {
return nil
}
if event.Privacy.OwnerUserID != userID || event.Privacy.Key == "" || len(event.Privacy.Rules) == 0 {
return domain.ErrPrivacyRuleInvalid
}
raw, err := json.Marshal(event.Privacy)
if err != nil {
return fmt.Errorf("encode privacy update payload: %w", err)
}
if _, err := db.Exec(ctx, `
INSERT INTO user_update_privacy_payloads (user_id, pts, payload)
VALUES ($1, $2, $3::jsonb)
`, userID, event.Pts, string(raw)); err != nil {
return fmt.Errorf("save privacy update payload: %w", err)
}
return nil
}
func appendQuickReplyPayload(ctx context.Context, db sqlcgen.DBTX, userID int64, event domain.UpdateEvent) error {
switch event.Type {
case domain.UpdateEventQuickReplies,
@ -466,6 +489,9 @@ func (s *UpdateEventStore) ListAfter(ctx context.Context, userID int64, pts, lim
}
out = append(out, event)
}
if err := s.hydratePrivacyEvents(ctx, out); err != nil {
return nil, err
}
return out, nil
}
@ -665,9 +691,77 @@ func (s *UpdateEventStore) BatchByCursor(ctx context.Context, cursors []store.Ev
}
out = append(out, event)
}
if err := s.hydratePrivacyEvents(ctx, out); err != nil {
return nil, err
}
return out, nil
}
type privacyEventCursor struct {
userID int64
pts int
}
// hydratePrivacyEvents fetches all immutable privacy payloads for one
// difference/outbox batch in one query. Ordinary event batches incur no extra
// query at all.
func (s *UpdateEventStore) hydratePrivacyEvents(ctx context.Context, events []domain.UpdateEvent) error {
indexes := make(map[privacyEventCursor]int)
userIDs := make([]int64, 0)
pts := make([]int32, 0)
for i := range events {
if events[i].Type != domain.UpdateEventPrivacy {
continue
}
key := privacyEventCursor{userID: events[i].UserID, pts: events[i].Pts}
indexes[key] = i
userIDs = append(userIDs, key.userID)
pts = append(pts, int32(key.pts))
}
if len(indexes) == 0 {
return nil
}
rows, err := s.db.Query(ctx, `
SELECT p.user_id, p.pts, p.payload::text
FROM unnest($1::bigint[], $2::int[]) AS requested(user_id, pts)
JOIN user_update_privacy_payloads p USING (user_id, pts)
`, userIDs, pts)
if err != nil {
return fmt.Errorf("list privacy update payloads: %w", err)
}
defer rows.Close()
found := 0
for rows.Next() {
var userID int64
var eventPts int
var raw string
if err := rows.Scan(&userID, &eventPts, &raw); err != nil {
return fmt.Errorf("scan privacy update payload: %w", err)
}
index, ok := indexes[privacyEventCursor{userID: userID, pts: eventPts}]
if !ok {
continue
}
var payload domain.PrivacyRules
if err := json.Unmarshal([]byte(raw), &payload); err != nil {
return fmt.Errorf("decode privacy update payload: %w", err)
}
if payload.OwnerUserID != userID || payload.Key == "" || len(payload.Rules) == 0 {
return fmt.Errorf("invalid privacy update payload for user %d pts %d", userID, eventPts)
}
events[index].Privacy = payload
delete(indexes, privacyEventCursor{userID: userID, pts: eventPts})
found++
}
if err := rows.Err(); err != nil {
return fmt.Errorf("list privacy update payloads rows: %w", err)
}
if found != len(userIDs) || len(indexes) != 0 {
return fmt.Errorf("privacy update payload missing")
}
return nil
}
func usersFromUpdateEventRow(row sqlcgen.ListUserUpdateEventsAfterRow) []domain.User {
return mergeEventUsers(
domain.User{

View file

@ -12,3 +12,16 @@ type PrivacyStore interface {
SetPrivacyRules(ctx context.Context, rules domain.PrivacyRules) error
ListPrivacyRules(ctx context.Context, ownerUserIDs []int64, keys []domain.PrivacyKey) ([]domain.PrivacyRules, error)
}
// PrivacyUpdateStore atomically commits an absolute privacy rule snapshot,
// allocates account pts, appends its durable event, and enqueues online
// dispatch. Implementations return the event with its final pts.
type PrivacyUpdateStore interface {
SetPrivacyRulesWithUpdate(
ctx context.Context,
rules domain.PrivacyRules,
event domain.UpdateEvent,
excludeAuthKeyID [8]byte,
excludeSessionID int64,
) (domain.UpdateEvent, error)
}