queueInflightExpiry enqueues a task to handle inflight expiry for a transaction. Parameters: - transactionID string: The ID of the transaction. - expiresAt time.Time: The expiration time for the inflight status. Returns: - error: An error if the task could not be enqueued.
(transactionID string, expiresAt time.Time)
| 79 | // Returns: |
| 80 | // - error: An error if the task could not be enqueued. |
| 81 | func (q *Queue) queueInflightExpiry(transactionID string, expiresAt time.Time) error { |
| 82 | IPayload, err := json.Marshal(transactionID) |
| 83 | if err != nil { |
| 84 | return err |
| 85 | } |
| 86 | taskOptions := []asynq.Option{ |
| 87 | asynq.TaskID(transactionID), |
| 88 | asynq.Queue(q.config.Queue.InflightExpiryQueue), |
| 89 | asynq.ProcessIn(time.Until(expiresAt)), |
| 90 | } |
| 91 | task := asynq.NewTask(q.config.Queue.InflightExpiryQueue, IPayload, taskOptions...) |
| 92 | _, err = q.Client.Enqueue(task) |
| 93 | if err != nil { |
| 94 | logrus.WithError(err).WithField("transaction_id", transactionID).Error("failed to enqueue inflight expiry") |
| 95 | return err |
| 96 | } |
| 97 | logrus.WithField("transaction_id", transactionID).Debug("successfully enqueued inflight expiry") |
| 98 | return nil |
| 99 | } |
| 100 | |
| 101 | // queueIndexBatch enqueues a batch of items to be indexed in dependency order. |
| 102 | // This ensures that dependencies (e.g., balances) are indexed before the primary item (e.g., transaction). |
no test coverage detected