Something went wrong. Try again.
collection of golang services under the Red Dwarf umbrella forked from whey.party/red-dwarf-server
Something went wrong. Try again.
2.7 kB · 113 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114package backstream
import ( "crypto/rand" "encoding/hex" "log" "sync" "time")
type Session struct { Ticket string Params BackfillParams LastAccessed time.Time
ListReposCursor string // Cursor for listReposByCollection if wantedDids=* CompletedDIDs map[string]bool // Set of DIDs that have been fully processed. listRecordsCursors map[string]string // Key: "did/collection", Value: cursor
mu sync.Mutex}
func (s *Session) GetListRecordsCursor(did, collection string) string { s.mu.Lock() defer s.mu.Unlock() key := did + "/" + collection return s.listRecordsCursors[key]}
func (s *Session) SetListRecordsCursor(did, collection, cursor string) { s.mu.Lock() defer s.mu.Unlock() key := did + "/" + collection s.listRecordsCursors[key] = cursor s.LastAccessed = time.Now()}
func (s *Session) MarkDIDComplete(did string) { s.mu.Lock() defer s.mu.Unlock() s.CompletedDIDs[did] = true s.LastAccessed = time.Now()}
func (s *Session) IsDIDComplete(did string) bool { s.mu.Lock() defer s.mu.Unlock() return s.CompletedDIDs[did]}
type SessionManager struct { sessions map[string]*Session ttl time.Duration mu sync.Mutex}
func NewSessionManager(ttl time.Duration) *SessionManager { sm := &SessionManager{ sessions: make(map[string]*Session), ttl: ttl, } go sm.cleanupLoop() return sm}
func (sm *SessionManager) GetOrCreate(ticket string, params BackfillParams) *Session { sm.mu.Lock() defer sm.mu.Unlock()
if session, exists := sm.sessions[ticket]; exists { log.Printf("Resuming existing session for ticket: %s", ticket) session.LastAccessed = time.Now() if session.CompletedDIDs == nil { session.CompletedDIDs = make(map[string]bool) } return session }
log.Printf("Creating new session for ticket: %s", ticket) newSession := &Session{ Ticket: ticket, Params: params, LastAccessed: time.Now(), listRecordsCursors: make(map[string]string), CompletedDIDs: make(map[string]bool), } sm.sessions[ticket] = newSession return newSession}
func (sm *SessionManager) cleanupLoop() { ticker := time.NewTicker(sm.ttl / 2) defer ticker.Stop() for range ticker.C { sm.mu.Lock() now := time.Now() for ticket, session := range sm.sessions { if now.Sub(session.LastAccessed) > sm.ttl { log.Printf("Session %s expired. Cleaning up.", ticket) delete(sm.sessions, ticket) } } sm.mu.Unlock() }}
func generateTicket() string { bytes := make([]byte, 16) if _, err := rand.Read(bytes); err != nil { return "fallback-ticket-" + time.Now().String() } return hex.EncodeToString(bytes)}