func (d *dlqMessageHandlerImpl) Purge(
	ctx context.Context,
	lastMessageID int64,
) error {

	ackLevel, err := d.replicationQueue.GetDLQAckLevel(ctx)
	if err != nil {
		return err
	}

	if err := d.replicationQueue.RangeDeleteMessagesFromDLQ(
		ctx,
		ackLevel,
		lastMessageID,
	); err != nil {
		return err
	}

	if err := d.replicationQueue.UpdateDLQAckLevel(
		ctx,
		lastMessageID,
	); err != nil {
		d.logger.Error("Failed to update DLQ ack level after purging messages", tag.Error(err))
	}

	return nil
}
// PurgeMessages purges domain replication DLQ messages
PurgeMessages purges domain replication DLQ messages
// PurgeMessages purges domain replication DLQ messages
func (d *dlqMessageHandlerImpl) Purge(
	ctx context.Context,
	lastMessageID int64,
) error {
	ackLevel, err := d.replicationQueue.GetDLQAckLevel(ctx)
	if err != nil {
		return err
	}
	if err := d.replicationQueue.RangeDeleteMessagesFromDLQ(
		ctx,
		ackLevel,
		lastMessageID,
	); err != nil {
		return err
	}
	if err := d.replicationQueue.UpdateDLQAckLevel(
		ctx,
		lastMessageID,
	); err != nil {
		d.logger.Error("Failed to update DLQ ack level after purging messages", tag.Error(err))
	}
	return nil
}
func ( d * dlqMessageHandlerImpl ) Purge ( ctx context . Context , lastMessageID int64 , ) error { ackLevel , err := d . replicationQueue . GetDLQAckLevel ( ctx ) if err != nil { return err } if err := d . replicationQueue . RangeDeleteMessagesFromDLQ ( ctx , ackLevel , lastMessageID , ) ; err != nil { return err } if err := d . replicationQueue . UpdateDLQAckLevel ( ctx , lastMessageID , ) ; err != nil { d . logger . Error ( " " , tag . Error ( err ) ) } return nil }
PurgeMessages purges domain replication DLQ messages
PurgeMessages purges domain replication DLQ messages