wangxuming/XQKqueue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add historical data management

Browse Source
This commit is contained in:
wangxuming
2026-07-22 10:18:49 +08:00
parent 66951b4dc3
commit c6044d972c
23 changed files with 3389 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
server/internal/database/database_integration_test.go
View File

@@ -45,8 +45,8 @@ func TestPostgresMigrationAndMaintenanceIntegration(t *testing.T) {
if err := sqlDB.QueryRowContext(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil {
t.Fatal(err)
}
if migrationCount < 12 {
t.Fatalf("migration count = %d, want at least 12", migrationCount)
if migrationCount < 14 {
t.Fatalf("migration count = %d, want at least 14", migrationCount)
}
var experiencedPeopleStartColumn string
if err := sqlDB.QueryRowContext(ctx, `
@@ -78,6 +78,16 @@ func TestPostgresMigrationAndMaintenanceIntegration(t *testing.T) {
if nullable != "YES" {
t.Fatalf("phone_hmac is_nullable = %q, want YES after purge migration", nullable)
}
var historyAnonymizedNullable string
if err := sqlDB.QueryRowContext(ctx, `
SELECT is_nullable
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = 'queue_tickets' AND column_name = 'history_anonymized_at'`).Scan(&historyAnonymizedNullable); err != nil {
t.Fatal(err)
}
if historyAnonymizedNullable != "YES" {
t.Fatalf("history_anonymized_at is_nullable = %q, want YES", historyAnonymizedNullable)
}
if _, err := sqlDB.ExecContext(ctx, `UPDATE users SET password_hash = $1, updated_at = now() WHERE username = $2`, "rotated-hash", "xqkwljtadmin"); err != nil {
t.Fatalf("protected super admin password rotation should be allowed: %v", err)
}
@@ -90,7 +100,7 @@ func TestPostgresMigrationAndMaintenanceIntegration(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if stats.PersonalDataPurged != 1 || stats.IdempotencyDeleted != 1 || stats.SessionsDeleted != 1 || stats.AuditDeleted != 1 {
if stats.PersonalDataPurged != 1 || stats.HistoryAnonymized != 1 || stats.IdempotencyDeleted != 1 || stats.SessionsDeleted != 1 || stats.AuditDeleted != 1 {
t.Fatalf("maintenance stats = %#v, want one row in each category", stats)
}
@@ -104,18 +114,32 @@ func TestPostgresMigrationAndMaintenanceIntegration(t *testing.T) {
if phoneCiphertext != nil || phoneNonce != nil || phoneHMAC != nil || purgedAt == nil {
t.Fatalf("purged ticket fields = %#v/%#v/%#v/%#v", phoneCiphertext, phoneNonce, phoneHMAC, purgedAt)
}
var historyPhoneCiphertext, historyPhoneNonce, historyPhoneHMAC, historyLastNameCiphertext, historyLastNameNonce, historyPurgedAt, historyAnonymizedAt any
var historyHonorific string
if err := sqlDB.QueryRowContext(ctx, `
SELECT phone_ciphertext, phone_nonce, phone_hmac, last_name_ciphertext, last_name_nonce,
personal_data_purged_at, history_anonymized_at, honorific
FROM queue_tickets WHERE id = $1`, fixture.historyTicketID).
Scan(&historyPhoneCiphertext, &historyPhoneNonce, &historyPhoneHMAC, &historyLastNameCiphertext, &historyLastNameNonce, &historyPurgedAt, &historyAnonymizedAt, &historyHonorific); err != nil {
t.Fatal(err)
}
if historyPhoneCiphertext != nil || historyPhoneNonce != nil || historyPhoneHMAC != nil || historyLastNameCiphertext != nil || historyLastNameNonce != nil || historyPurgedAt == nil || historyAnonymizedAt == nil || historyHonorific != "游客" {
t.Fatalf("historical anonymization fields = %#v/%#v/%#v/%#v/%#v/%#v/%#v/%q", historyPhoneCiphertext, historyPhoneNonce, historyPhoneHMAC, historyLastNameCiphertext, historyLastNameNonce, historyPurgedAt, historyAnonymizedAt, historyHonorific)
}
}
type maintenanceFixture struct {
userID string
projectID string
ticketID string
userID string
projectID string
ticketID string
historyTicketID string
}
func newMaintenanceFixture(t *testing.T, ctx context.Context, db *sql.DB) maintenanceFixture {
t.Helper()
now := time.Now().UTC().Truncate(time.Microsecond)
userID, projectID, sessionID, ticketID := uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString()
userID, projectID, sessionID, ticketID, historyTicketID := uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString()
code := "MNT" + strings.ToUpper(strings.ReplaceAll(userID[:5], "-", ""))
if _, err := db.ExecContext(ctx, `
INSERT INTO users (id, username, password_hash, role, active)
@@ -145,6 +169,16 @@ func newMaintenanceFixture(t *testing.T, ctx context.Context, db *sql.DB) mainte
$8, $8, $9, $10)`, ticketID, projectID, sessionID, strings.Repeat("a", 64), []byte("ciphertext"), make([]byte, 12), strings.Repeat("b", 64), now.Add(-72*time.Hour), now.Add(-time.Hour), userID); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `
INSERT INTO queue_tickets (
id, project_id, queue_session_id, ticket_number, display_number,
public_token_hash, phone_ciphertext, phone_nonce, phone_hmac,
last_name_ciphertext, last_name_nonce, honorific, status, joined_at,
completed_at, personal_data_purge_at, created_by
) VALUES ($1, $2, $3, 2, '00002', $4, $5, $6, $7, $8, $9, '女士', 'COMPLETED',
$10, $10, $11, $12)`, historyTicketID, projectID, sessionID, strings.Repeat("e", 64), []byte("history-ciphertext"), make([]byte, 12), strings.Repeat("f", 64), []byte("history-last-name"), make([]byte, 12), now.Add(-100*24*time.Hour), now.Add(30*24*time.Hour), userID); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(ctx, `
INSERT INTO idempotency_keys (project_id, user_id, scope, key, request_hash, response_code, response_body, expires_at)
VALUES ($1, $2, 'TEST', $3, $4, 200, '{}', $5)`, projectID, userID, "expired-"+userID, strings.Repeat("c", 64), now.Add(-time.Hour)); err != nil {
@@ -160,5 +194,5 @@ func newMaintenanceFixture(t *testing.T, ctx context.Context, db *sql.DB) mainte
VALUES ($1, $2, 'TEST', 'TEST', '{}', $3, 'integration-test', $4)`, projectID, userID, "request-"+userID, now.Add(-time.Hour)); err != nil {
t.Fatal(err)
}
return maintenanceFixture{userID: userID, projectID: projectID, ticketID: ticketID}
return maintenanceFixture{userID: userID, projectID: projectID, ticketID: ticketID, historyTicketID: historyTicketID}
}

29
server/internal/database/maintenance.go
View File

@@ -15,6 +15,7 @@ const maintenanceLockID int64 = 733081337591911
// transaction while trying to clean an entire historical table.
type CleanupStats struct {
PersonalDataPurged int64
HistoryAnonymized int64
IdempotencyDeleted int64
SessionsDeleted int64
AuditDeleted int64
@@ -83,6 +84,31 @@ func RunMaintenanceOnce(ctx context.Context, db *sql.DB, now time.Time, batchSiz
if err != nil {
return rollback(fmt.Errorf("purge personal data: %w", err))
}
stats.HistoryAnonymized, err = boundedUpdate(ctx, tx, `
WITH candidates AS (
SELECT id
FROM queue_tickets
WHERE history_anonymized_at IS NULL
AND joined_at <= ($1::timestamptz - interval '90 days')
ORDER BY joined_at, id
LIMIT $2
FOR UPDATE SKIP LOCKED
)
UPDATE queue_tickets AS ticket
SET phone_ciphertext = NULL,
phone_nonce = NULL,
phone_hmac = NULL,
last_name_ciphertext = NULL,
last_name_nonce = NULL,
honorific = '游客',
personal_data_purged_at = COALESCE(ticket.personal_data_purged_at, $1),
history_anonymized_at = $1,
updated_at = $1
FROM candidates
WHERE ticket.id = candidates.id`, now, batchSize)
if err != nil {
return rollback(fmt.Errorf("mark historical data anonymized: %w", err))
}
stats.IdempotencyDeleted, err = boundedDelete(ctx, tx, `
WITH candidates AS (
SELECT id
@@ -160,9 +186,10 @@ func StartMaintenance(ctx context.Context, db *sql.DB, logger *slog.Logger, inte
logger.Error("database maintenance failed", "error", err)
return
}
if stats.PersonalDataPurged+stats.IdempotencyDeleted+stats.SessionsDeleted+stats.AuditDeleted > 0 {
if stats.PersonalDataPurged+stats.HistoryAnonymized+stats.IdempotencyDeleted+stats.SessionsDeleted+stats.AuditDeleted > 0 {
logger.Info("database maintenance completed",
"personal_data_purged", stats.PersonalDataPurged,
"history_anonymized", stats.HistoryAnonymized,
"idempotency_deleted", stats.IdempotencyDeleted,
"sessions_deleted", stats.SessionsDeleted,
"audit_deleted", stats.AuditDeleted,