lemintare
6b189bfce4
This commit introduces a new feature for global role synchronization in Discord, allowing for periodic reconciliation of managed roles. A new environment variable, `DISCORD_GLOBAL_SYNC_INTERVAL`, has been added to configure the synchronization interval, defaulting to 5 minutes. The `RoleWorker` has been updated to schedule global sync jobs, ensuring that missing managed roles are restored and extra assignments are removed without affecting unrelated server roles. Database schema changes support the new synchronization logic, and tests have been added to validate the functionality of the global reconciliation process.
322 lines
11 KiB
Go
322 lines
11 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"mixmaker/backend/internal/application"
|
|
"mixmaker/backend/internal/domain"
|
|
)
|
|
|
|
func (s *Store) EnqueueDiscordRoleSync(ctx context.Context, eventID, action string, generation int64) error {
|
|
_, err := s.pool.Exec(ctx, `INSERT INTO discord_role_sync_jobs(event_id,action,generation,role_snapshot)
|
|
SELECT $1,$2,$3,COALESCE(jsonb_agg(discord_role_id) FILTER (WHERE discord_role_id IS NOT NULL),'[]'::jsonb)
|
|
FROM discord_managed_roles WHERE scope='event' AND event_id=$1
|
|
ON CONFLICT(event_id,action,generation) DO UPDATE
|
|
SET status='pending',available_at=now(),updated_at=now(),completed_at=NULL`,
|
|
eventID, action, generation)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) SeedDiscordRoleSyncJobs(ctx context.Context) error {
|
|
if _, err := s.pool.Exec(ctx, `INSERT INTO discord_role_sync_jobs(event_id,action,generation)
|
|
SELECT e.id,'reconcile',e.version
|
|
FROM events e
|
|
JOIN event_rosters er ON er.event_id=e.id
|
|
WHERE e.state IN ('RostersConfirmed','BracketDraft','Live')
|
|
AND COALESCE((er.body->>'confirmed')::boolean,false)=true
|
|
ON CONFLICT(event_id,action,generation) DO NOTHING`); err != nil {
|
|
return err
|
|
}
|
|
_, err := s.pool.Exec(ctx, `INSERT INTO discord_role_sync_jobs(event_id,action,generation)
|
|
SELECT DISTINCT e.id,'rsvp',e.version
|
|
FROM events e JOIN rsvps r ON r.event_id=e.id
|
|
WHERE e.state NOT IN ('Completed','Cancelled')
|
|
ON CONFLICT(event_id,action,generation) DO NOTHING`)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) ScheduleGlobalDiscordRoleSync(ctx context.Context, bucket int64) error {
|
|
_, err := s.pool.Exec(ctx, `INSERT INTO discord_role_sync_jobs(event_id,action,generation)
|
|
VALUES('__global__','full_reconcile',$1)
|
|
ON CONFLICT(event_id,action,generation) DO NOTHING`, bucket)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) ClaimDiscordRoleSyncJob(ctx context.Context) (application.DiscordRoleSyncJob, error) {
|
|
var job application.DiscordRoleSyncJob
|
|
var snapshot []byte
|
|
err := s.pool.QueryRow(ctx, `WITH candidate AS (
|
|
SELECT id FROM discord_role_sync_jobs
|
|
WHERE (status='pending' AND available_at<=now())
|
|
OR (status='processing' AND updated_at<now()-interval '5 minutes')
|
|
ORDER BY id
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 1
|
|
)
|
|
UPDATE discord_role_sync_jobs j
|
|
SET status='processing',attempts=j.attempts+1,updated_at=now()
|
|
FROM candidate
|
|
WHERE j.id=candidate.id
|
|
RETURNING j.id,j.event_id,j.action,j.generation,j.role_snapshot,j.attempts`).
|
|
Scan(&job.ID, &job.EventID, &job.Action, &job.Generation, &snapshot, &job.Attempts)
|
|
if err != nil {
|
|
return job, mapError(err)
|
|
}
|
|
if err = json.Unmarshal(snapshot, &job.RoleSnapshot); err != nil {
|
|
return job, err
|
|
}
|
|
return job, nil
|
|
}
|
|
|
|
func (s *Store) CompleteDiscordRoleSyncJob(ctx context.Context, jobID int64, warning string) error {
|
|
_, err := s.pool.Exec(ctx, `UPDATE discord_role_sync_jobs
|
|
SET status='completed',last_error=$2,completed_at=now(),updated_at=now()
|
|
WHERE id=$1`, jobID, warning)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) RetryDiscordRoleSyncJob(ctx context.Context, jobID int64, message string, availableAt time.Time) error {
|
|
_, err := s.pool.Exec(ctx, `UPDATE discord_role_sync_jobs
|
|
SET status='pending',last_error=$2,available_at=$3,updated_at=now()
|
|
WHERE id=$1`, jobID, message, availableAt)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) ListDiscordManagedRoles(ctx context.Context, eventID string) ([]application.DiscordManagedRole, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT scope,event_id,team_id,kind,discord_role_id,role_name,hoist
|
|
FROM discord_managed_roles
|
|
WHERE scope='global' OR event_id=$1
|
|
ORDER BY scope,event_id,team_id,kind`, eventID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
roles := make([]application.DiscordManagedRole, 0)
|
|
for rows.Next() {
|
|
var role application.DiscordManagedRole
|
|
if err = rows.Scan(&role.Scope, &role.EventID, &role.TeamID, &role.Kind, &role.DiscordRoleID, &role.RoleName, &role.Hoist); err != nil {
|
|
return nil, err
|
|
}
|
|
roles = append(roles, role)
|
|
}
|
|
return roles, rows.Err()
|
|
}
|
|
|
|
func (s *Store) ListAllDiscordManagedRoles(ctx context.Context) ([]application.DiscordManagedRole, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT scope,event_id,team_id,kind,discord_role_id,role_name,hoist
|
|
FROM discord_managed_roles ORDER BY scope,event_id,team_id,kind`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
roles := make([]application.DiscordManagedRole, 0)
|
|
for rows.Next() {
|
|
var role application.DiscordManagedRole
|
|
if err = rows.Scan(&role.Scope, &role.EventID, &role.TeamID, &role.Kind, &role.DiscordRoleID, &role.RoleName, &role.Hoist); err != nil {
|
|
return nil, err
|
|
}
|
|
roles = append(roles, role)
|
|
}
|
|
return roles, rows.Err()
|
|
}
|
|
|
|
func (s *Store) UpsertDiscordManagedRole(ctx context.Context, role application.DiscordManagedRole) error {
|
|
_, err := s.pool.Exec(ctx, `INSERT INTO discord_managed_roles(scope,event_id,team_id,kind,discord_role_id,role_name,hoist)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7)
|
|
ON CONFLICT(scope,event_id,team_id,kind) DO UPDATE
|
|
SET discord_role_id=excluded.discord_role_id,role_name=excluded.role_name,hoist=excluded.hoist,updated_at=now()`,
|
|
role.Scope, role.EventID, role.TeamID, role.Kind, role.DiscordRoleID, role.RoleName, role.Hoist)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) DeleteDiscordManagedRole(ctx context.Context, role application.DiscordManagedRole) error {
|
|
_, err := s.pool.Exec(ctx, `DELETE FROM discord_managed_roles
|
|
WHERE scope=$1 AND event_id=$2 AND team_id=$3 AND kind=$4`,
|
|
role.Scope, role.EventID, role.TeamID, role.Kind)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) ListDiscordRoleAssignments(ctx context.Context, discordRoleID string) ([]string, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT discord_user_id FROM discord_role_assignments
|
|
WHERE discord_role_id=$1 ORDER BY discord_user_id`, discordRoleID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
userIDs := make([]string, 0)
|
|
for rows.Next() {
|
|
var userID string
|
|
if err = rows.Scan(&userID); err != nil {
|
|
return nil, err
|
|
}
|
|
userIDs = append(userIDs, userID)
|
|
}
|
|
return userIDs, rows.Err()
|
|
}
|
|
|
|
func (s *Store) UpsertDiscordRoleAssignment(ctx context.Context, discordRoleID, discordUserID string) error {
|
|
_, err := s.pool.Exec(ctx, `INSERT INTO discord_role_assignments(discord_role_id,discord_user_id)
|
|
VALUES($1,$2) ON CONFLICT(discord_role_id,discord_user_id) DO UPDATE SET updated_at=now()`,
|
|
discordRoleID, discordUserID)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) DeleteDiscordRoleAssignment(ctx context.Context, discordRoleID, discordUserID string) error {
|
|
_, err := s.pool.Exec(ctx, `DELETE FROM discord_role_assignments
|
|
WHERE discord_role_id=$1 AND discord_user_id=$2`, discordRoleID, discordUserID)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) SetDiscordRoleAssignments(ctx context.Context, discordRoleID string, discordUserIDs []string) error {
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
if _, err = tx.Exec(ctx, `DELETE FROM discord_role_assignments WHERE discord_role_id=$1`, discordRoleID); err != nil {
|
|
return err
|
|
}
|
|
for _, userID := range discordUserIDs {
|
|
if _, err = tx.Exec(ctx, `INSERT INTO discord_role_assignments(discord_role_id,discord_user_id)
|
|
VALUES($1,$2)`, discordRoleID, userID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (s *Store) GetDiscordRoleRoster(ctx context.Context, eventID string) (application.DiscordRoleRoster, error) {
|
|
roster, err := s.GetRoster(ctx, eventID)
|
|
if err != nil {
|
|
return application.DiscordRoleRoster{}, err
|
|
}
|
|
return s.discordRoleRoster(ctx, roster)
|
|
}
|
|
|
|
func (s *Store) ListActiveDiscordRoleRosters(ctx context.Context) ([]application.DiscordRoleRoster, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT er.body
|
|
FROM event_rosters er
|
|
JOIN events e ON e.id=er.event_id
|
|
WHERE e.state IN ('RostersConfirmed','BracketDraft','Live')
|
|
AND COALESCE((er.body->>'confirmed')::boolean,false)=true
|
|
ORDER BY er.event_id`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
rosters := make([]domain.RosterDraft, 0)
|
|
for rows.Next() {
|
|
var body []byte
|
|
var roster domain.RosterDraft
|
|
if err = rows.Scan(&body); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = json.Unmarshal(body, &roster); err != nil {
|
|
return nil, err
|
|
}
|
|
rosters = append(rosters, roster)
|
|
}
|
|
if err = rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]application.DiscordRoleRoster, 0, len(rosters))
|
|
for _, roster := range rosters {
|
|
item, rosterErr := s.discordRoleRoster(ctx, roster)
|
|
if rosterErr != nil {
|
|
return nil, rosterErr
|
|
}
|
|
out = append(out, item)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Store) GetDiscordRoleRegistration(ctx context.Context, eventID string) (application.DiscordRoleRegistration, error) {
|
|
event, err := s.GetEvent(ctx, eventID)
|
|
if err != nil {
|
|
return application.DiscordRoleRegistration{}, err
|
|
}
|
|
registrations, err := s.ListRSVPs(ctx, eventID)
|
|
if err != nil {
|
|
return application.DiscordRoleRegistration{}, err
|
|
}
|
|
playerIDs := make([]string, 0, len(registrations))
|
|
for _, registration := range registrations {
|
|
playerIDs = append(playerIDs, registration.PlayerID)
|
|
}
|
|
discordIDs, err := s.playerDiscordIDs(ctx, playerIDs)
|
|
if err != nil {
|
|
return application.DiscordRoleRegistration{}, err
|
|
}
|
|
return application.DiscordRoleRegistration{Event: event, Registrations: registrations, PlayerDiscordIDs: discordIDs}, nil
|
|
}
|
|
|
|
func (s *Store) ListActiveDiscordRoleRegistrations(ctx context.Context) ([]application.DiscordRoleRegistration, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT id FROM events
|
|
WHERE state NOT IN ('Completed','Cancelled') ORDER BY id`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
eventIDs := make([]string, 0)
|
|
for rows.Next() {
|
|
var eventID string
|
|
if err = rows.Scan(&eventID); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
eventIDs = append(eventIDs, eventID)
|
|
}
|
|
rows.Close()
|
|
if err = rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]application.DiscordRoleRegistration, 0, len(eventIDs))
|
|
for _, eventID := range eventIDs {
|
|
registration, registrationErr := s.GetDiscordRoleRegistration(ctx, eventID)
|
|
if registrationErr != nil {
|
|
return nil, registrationErr
|
|
}
|
|
out = append(out, registration)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Store) discordRoleRoster(ctx context.Context, roster domain.RosterDraft) (application.DiscordRoleRoster, error) {
|
|
playerIDs := make([]string, 0)
|
|
for _, team := range roster.Teams {
|
|
for _, slot := range team.Slots {
|
|
if slot.PlayerID != "" {
|
|
playerIDs = append(playerIDs, slot.PlayerID)
|
|
}
|
|
}
|
|
}
|
|
discordIDs, err := s.playerDiscordIDs(ctx, playerIDs)
|
|
if err != nil {
|
|
return application.DiscordRoleRoster{}, err
|
|
}
|
|
return application.DiscordRoleRoster{Roster: roster, PlayerDiscordIDs: discordIDs}, nil
|
|
}
|
|
|
|
func (s *Store) playerDiscordIDs(ctx context.Context, playerIDs []string) (map[string]string, error) {
|
|
discordIDs := make(map[string]string)
|
|
if len(playerIDs) == 0 {
|
|
return discordIDs, nil
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT p.id,a.discord_id
|
|
FROM players p JOIN accounts a ON a.id=p.account_id
|
|
WHERE p.id=ANY($1)`, playerIDs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var playerID, discordID string
|
|
if err = rows.Scan(&playerID, &discordID); err != nil {
|
|
return nil, err
|
|
}
|
|
discordIDs[playerID] = discordID
|
|
}
|
|
return discordIDs, rows.Err()
|
|
}
|