mixmaker/backend/internal/adapter/postgres/discord_roles.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 {
	_, 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`)
	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
		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); 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)
		VALUES($1,$2,$3,$4,$5,$6)
		ON CONFLICT(scope,event_id,team_id,kind) DO UPDATE
		SET discord_role_id=excluded.discord_role_id,role_name=excluded.role_name,updated_at=now()`,
		role.Scope, role.EventID, role.TeamID, role.Kind, role.DiscordRoleID, role.RoleName)
	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) 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) 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 := make(map[string]string)
	if len(playerIDs) > 0 {
		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 application.DiscordRoleRoster{}, err
		}
		defer rows.Close()
		for rows.Next() {
			var playerID, discordID string
			if err = rows.Scan(&playerID, &discordID); err != nil {
				return application.DiscordRoleRoster{}, err
			}
			discordIDs[playerID] = discordID
		}
		if err = rows.Err(); err != nil {
			return application.DiscordRoleRoster{}, err
		}
	}
	return application.DiscordRoleRoster{Roster: roster, PlayerDiscordIDs: discordIDs}, nil
}