purge expired tokens

manager/manager.go

@@ -31,6 +31,7 @@ const (
 	ResponseBufferCap  = 100
 	TokenLength        = 12
 	TokenValidityDays  = 7
+	TokenPurgeInterval = time.Hour
 )
 
 type event struct {
@@ -53,6 +54,7 @@ func startManager(pipe model.Pipe, zmqConf model.ZeromqServerInfo, storageClient
 		return nil, fmt.Errorf("error creating keys for CA: %s", err)
 	}
 
+	go m.purgeExpiredTokens()
 	go m.manageResponses()
 	return m, nil
 }
@@ -432,6 +434,19 @@ func (m *manager) getServerInfo() (*model.ServerInfo, error) {
 	}, nil
 }
 
+func (m *manager) purgeExpiredTokens() {
+	for t := time.Now(); true; t = <-time.Tick(TokenPurgeInterval) {
+		total, err := m.storage.PurgeOldTokens(t.AddDate(0, 0, -TokenValidityDays))
+		if err != nil {
+			log.Printf("Error purging tokens: %s", err)
+			continue
+		}
+		if total > 0 {
+			log.Printf("Purged %d old tokens.", total)
+		}
+	}
+}
+
 type tokenSet struct {
 	Name      string             `json:"name"`
 	Available int                `json:"available"`

manager/storage/storage.go

@@ -41,6 +41,7 @@ type Storage interface {
 	findToken(hash string) (found bool, err error)
 	DeleteTokenTrans(hash string) (found bool, trans *transaction, err error)
 	DeleteTokens(name string) error
+	PurgeOldTokens(to time.Time) (total int64, err error)
 	//
 	DoBulk(...interface{}) error
 }
@@ -734,6 +735,16 @@ func (s *storage) DeleteTokens(name string) error {
 	return nil
 }
 
+func (s *storage) PurgeOldTokens(to time.Time) (int64, error) {
+	query := elastic.NewBoolQuery().Filter(elastic.NewRangeQuery("expiresAt").To(to))
+
+	res, err := s.client.DeleteByQuery(indexToken).Query(query).Do(s.ctx)
+	if err != nil {
+		return 0, err
+	}
+	return res.Total, nil
+}
+
 // DoBulk performs elastic.BulkableRequests
 func (s *storage) DoBulk(requests ...interface{}) error {
 	bulk := s.client.Bulk()