migration

var V1MultitenancyMigration = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		cidOwners, err := v0CidOwners(ds)
		if err != nil {
			return fmt.Errorf("getting cid owners: %s", err)
		}
		log.Infof("Starting job logger migration...")
		if err := migrateJobLogger(ds, cidOwners); err != nil {
			return fmt.Errorf("migrating job logger: %s", err)
		}
		log.Infof("Job logger migration finished")

		log.Infof("Starting storage info migration...")
		if err := migrateStorageInfo(ds, cidOwners); err != nil {
			return fmt.Errorf("migrating storage info: %s", err)
		}
		log.Infof("Storage info migration finished")

		log.Infof("Starting trackstore migration...")
		if err := migrateTrackstore(ds, cidOwners); err != nil {
			return fmt.Errorf("migrating trackstore: %s", err)
		}
		log.Infof("Trackstore migration finished")

		log.Infof("Starting started deals migration...")
		if err := migrateStartedDeals(ds); err != nil {
			return fmt.Errorf("migrating trackstore: %s", err)
		}
		log.Infof("Started deals migration finished")

		log.Infof("Starting pinstore filling migration...")
		if err := pinstoreFilling(ds, cidOwners); err != nil {
			return fmt.Errorf("filling pinstore: %s", err)
		}
		log.Infof("Pinstore filling migration finished")

		return nil
	},
}

var V2StorageInfoDealIDs = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		propDealIDMap, err := v2GenProposalCidToDealID(ds)
		if err != nil {
			return fmt.Errorf("getting cid owners: %s", err)
		}
		log.Infof("Starting DealID filling...")
		if err := v2MigrationDealIDFilling(ds, propDealIDMap); err != nil {
			return fmt.Errorf("migrating job logger: %s", err)
		}
		log.Infof("DealID filling migration finished")

		return nil
	},
}

var V3StorageJobsIndexMigration = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		q := query.Query{Prefix: "/ffs/scheduler/sjstore/job"}
		res, err := ds.Query(q)
		if err != nil {
			return fmt.Errorf("querying sjstore jobs: %s", err)
		}
		defer func() { _ = res.Close() }()

		var count int
		var lock sync.Mutex
		var errors []string
		lim := make(chan struct{}, 1000)
		for r := range res.Next() {
			if r.Error != nil {
				return fmt.Errorf("iterating results: %s", r.Error)
			}

			lim <- struct{}{}

			r := r
			go func() {
				defer func() { <-lim }()

				var job storageJob
				if err := json.Unmarshal(r.Value, &job); err != nil {
					lock.Lock()
					errors = append(errors, fmt.Sprintf("unmarshaling job: %s", err))
					lock.Unlock()
					return
				}

				apiidKey := datastore.NewKey("/ffs/scheduler/sjstore/apiid").ChildString(job.APIID).ChildString(job.Cid.String()).ChildString(fmt.Sprintf("%d", job.CreatedAt))
				cidKey := datastore.NewKey("/ffs/scheduler/sjstore/cid").ChildString(job.Cid.String()).ChildString(job.APIID).ChildString(fmt.Sprintf("%d", job.CreatedAt))

				if err := ds.Put(apiidKey, []byte(job.ID)); err != nil {
					lock.Lock()
					errors = append(errors, fmt.Sprintf("putting apiid index record in datastore: %s", err))
					lock.Unlock()
					return

				}
				if err := ds.Put(cidKey, []byte(job.ID)); err != nil {
					lock.Lock()
					errors = append(errors, fmt.Sprintf("putting cid index record in datastore: %s", err))
					lock.Unlock()
					return

				}

			}()
			count++
		}

		for i := 0; i < cap(lim); i++ {
			lim <- struct{}{}
		}

		if len(errors) > 0 {
			for _, m := range errors {
				log.Error(m)
			}
			return fmt.Errorf("migration had %d errors", len(errors))
		}

		log.Infof("migration indexed %d storage-jobs", count)

		return nil
	},
}

var V4RecordsMigration = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		if err := v4IncludeIDInRetrievalRecords(ds); err != nil {
			return fmt.Errorf("including ids in retrieval records: %s", err)
		}

		if err := v4PopulateUpdatedAtIndex(ds); err != nil {
			return fmt.Errorf("populating updated at index: %s", err)
		}

		return nil
	},
}

var V5DeleteOldMinerIndex = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		q := query.Query{Prefix: "/index/miner/chainstore"}
		res, err := ds.Query(q)
		if err != nil {
			return fmt.Errorf("querying records: %s", err)
		}
		defer func() { _ = res.Close() }()

		var count int
		for v := range res.Next() {
			if err := ds.Delete(datastore.NewKey(v.Key)); err != nil {
				return fmt.Errorf("deleting miner chainstore key: %s", err)
			}
			count++
		}
		log.Infof("deleted %d chainstore keys", count)

		return nil
	},
}