forked from baron/baron-sso
chan
31d107ff2e
- Added support for fixed UUIDs during bulk registration (Search-first + ExternalID mapping) - Implemented idempotency and visibility restoration for soft-deleted users - Enhanced bulk upload UI to show 'New/Updated/Unchanged' status and modified fields - Added logic to reclaim identifiers (login_id) from colliding records - Added frontend E2E and backend unit tests for UUID integrity and conflict handling - Fixed i18n, formatting, and mock tests to satisfy code-check - Applied 'go fix' for 'omitzero' tags and general Go standards
89 lines
2.8 KiB
Go
89 lines
2.8 KiB
Go
package repository
|
|
|
|
import (
|
|
"baron-sso-backend/internal/domain"
|
|
"context"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type KetoOutboxRepository interface {
|
|
Create(ctx context.Context, entry *domain.KetoOutbox) error
|
|
CreateWithTx(tx *gorm.DB, entry *domain.KetoOutbox) error
|
|
FindPending(ctx context.Context, limit int) ([]domain.KetoOutbox, error)
|
|
ListCurrentBySubject(ctx context.Context, namespace, subject string) ([]domain.KetoOutbox, error)
|
|
UpdateStatus(ctx context.Context, id string, status string, retryCount int, lastError string) error
|
|
MarkProcessed(ctx context.Context, id string) error
|
|
}
|
|
|
|
type ketoOutboxRepository struct {
|
|
db *gorm.DB
|
|
}
|
|
|
|
func NewKetoOutboxRepository(db *gorm.DB) KetoOutboxRepository {
|
|
return &ketoOutboxRepository{db: db}
|
|
}
|
|
|
|
func (r *ketoOutboxRepository) Create(ctx context.Context, entry *domain.KetoOutbox) error {
|
|
return r.db.WithContext(ctx).Create(entry).Error
|
|
}
|
|
|
|
func (r *ketoOutboxRepository) CreateWithTx(tx *gorm.DB, entry *domain.KetoOutbox) error {
|
|
return tx.Create(entry).Error
|
|
}
|
|
|
|
func (r *ketoOutboxRepository) FindPending(ctx context.Context, limit int) ([]domain.KetoOutbox, error) {
|
|
var entries []domain.KetoOutbox
|
|
err := r.db.WithContext(ctx).
|
|
Where("status = ?", domain.KetoOutboxStatusPending).
|
|
Order("created_at asc").
|
|
Limit(limit).
|
|
Find(&entries).Error
|
|
return entries, err
|
|
}
|
|
|
|
func (r *ketoOutboxRepository) ListCurrentBySubject(ctx context.Context, namespace, subject string) ([]domain.KetoOutbox, error) {
|
|
var entries []domain.KetoOutbox
|
|
if err := r.db.WithContext(ctx).
|
|
Where("namespace = ? AND subject = ? AND status <> ?", namespace, subject, domain.KetoOutboxStatusFailed).
|
|
Order("created_at desc").
|
|
Order("updated_at desc").
|
|
Find(&entries).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
current := make([]domain.KetoOutbox, 0, len(entries))
|
|
seen := make(map[string]struct{}, len(entries))
|
|
for _, entry := range entries {
|
|
key := entry.Namespace + "\x00" + entry.Object + "\x00" + entry.Relation + "\x00" + entry.Subject
|
|
if _, exists := seen[key]; exists {
|
|
continue
|
|
}
|
|
seen[key] = struct{}{}
|
|
if entry.Action == domain.KetoOutboxActionCreate {
|
|
current = append(current, entry)
|
|
}
|
|
}
|
|
|
|
return current, nil
|
|
}
|
|
|
|
func (r *ketoOutboxRepository) UpdateStatus(ctx context.Context, id string, status string, retryCount int, lastError string) error {
|
|
return r.db.WithContext(ctx).Model(&domain.KetoOutbox{}).Where("id = ?", id).Updates(map[string]any{
|
|
"status": status,
|
|
"retry_count": retryCount,
|
|
"last_error": lastError,
|
|
"updated_at": time.Now(),
|
|
}).Error
|
|
}
|
|
|
|
func (r *ketoOutboxRepository) MarkProcessed(ctx context.Context, id string) error {
|
|
now := time.Now()
|
|
return r.db.WithContext(ctx).Model(&domain.KetoOutbox{}).Where("id = ?", id).Updates(map[string]any{
|
|
"status": domain.KetoOutboxStatusProcessed,
|
|
"processed_at": &now,
|
|
"updated_at": now,
|
|
}).Error
|
|
}
|