45 lines
1.4 KiB
Go
45 lines
1.4 KiB
Go
package creator
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
)
|
|
|
|
func (s *Store) ClaimSourceSync(ctx context.Context, sourceType, sourceID string) (string, error) {
|
|
if sourceType != SourceOwned && sourceType != SourceCompetitor || sourceID == "" {
|
|
return "", ErrInvalid
|
|
}
|
|
token := newID("source-lease")
|
|
var claimed string
|
|
err := s.db.QueryRowContext(ctx, `
|
|
INSERT INTO creator_source_sync_lease (source_type,source_id,lease_token,lease_until,updated_at)
|
|
VALUES ($1,$2,$3,now() + interval '10 minutes',now())
|
|
ON CONFLICT (source_type,source_id) DO UPDATE SET lease_token=EXCLUDED.lease_token,
|
|
lease_until=EXCLUDED.lease_until,updated_at=now()
|
|
WHERE creator_source_sync_lease.lease_until <= now()
|
|
RETURNING lease_token`, sourceType, sourceID, token).Scan(&claimed)
|
|
if err == sql.ErrNoRows {
|
|
return "", ErrConflict
|
|
}
|
|
if err != nil {
|
|
return "", databaseError(err)
|
|
}
|
|
return claimed, nil
|
|
}
|
|
|
|
func (s *Store) ReleaseSourceSync(ctx context.Context, sourceType, sourceID, token string) error {
|
|
if sourceType != SourceOwned && sourceType != SourceCompetitor || sourceID == "" || token == "" {
|
|
return ErrInvalid
|
|
}
|
|
result, err := s.db.ExecContext(ctx, `DELETE FROM creator_source_sync_lease WHERE source_type=$1 AND source_id=$2 AND lease_token=$3`, sourceType, sourceID, token)
|
|
if err != nil {
|
|
return databaseError(err)
|
|
}
|
|
if rows, err := result.RowsAffected(); err != nil {
|
|
return err
|
|
} else if rows != 1 {
|
|
return ErrConflict
|
|
}
|
|
return nil
|
|
}
|