Method indexBatchData

cmd/workers.go:228–258

indexBatchData indexes a batch of items into TypeSense in dependency order. It first indexes all dependencies (e.g., balances), then indexes the primary item (e.g., transaction). This ensures referential integrity in the search index.

(ctx context.Context, t *asynq.Task)

// It first indexes all dependencies (e.g., balances), then indexes the primary item (e.g., transaction).
// This ensures referential integrity in the search index.
func (b *ledgerforgeInstance) indexBatchData(ctx context.Context, t *asynq.Task) error {
	if b.cnf.TypeSense.Dns == "" {
		return nil
	}

	var batch search.IndexBatch

	// Unmarshal the batch data from the task payload.
	if err := json.Unmarshal(t.Payload(), &batch); err != nil {
		logrus.Error(err)
		return err
	}

	// Initialize a new TypeSense client and ensure collections exist.
	newSearch := search.NewTypesenseClient(b.cnf.TypeSenseKey, []string{b.cnf.TypeSense.Dns})
	err := newSearch.EnsureCollectionsExist(ctx)
	if err != nil {
		logrus.Errorf("Failed to ensure collections exist: %v", err)
		return err
	}

	// Handle the batch notification - indexes dependencies first, then primary.
	err = newSearch.HandleBatchNotification(ctx, &batch)
	if err != nil {
		logrus.Errorf("Error indexing batch %s: %v", batch.ID, err)
		return err
	}

	logrus.Errorf(" [*] Batch indexed: %s (deps: %d)", batch.ID, len(batch.Dependencies))
	return nil
}

// processInflightExpiry handles the expiry of inflight transactions.
// It voids the transaction by its ID and logs the action.