public static class AutoShipStreamBatching
   {
       public static void Run()
       {
           ShipLink.Helpers.LogHelper.LogMessage(
               "AutoShipStream: batching loop started (every " + (AutoShipStreamConfig.BatchIntervalMs / 1000) +
               "s, up to " + AutoShipStreamConfig.BatchSize + " records per batch).");

           DateTime nextSweep = DateTime.UtcNow;

           try
           {
               while (!AutoShipStreamControl.ShouldStop())
               {
                   try
                   {
                       if (DateTime.UtcNow >= nextSweep)
                       {
                           SweepDeadlines();
                           nextSweep = DateTime.UtcNow.AddMilliseconds(AutoShipStreamConfig.DeadlineSweepMs);
                       }

                       ClaimBatch();
                   }
                   catch (Exception ex)
                   {
                       // Never let one bad tick kill the loop - the feed keeps arriving
                       // whether or not this thread is healthy.
                       ShipLink.Helpers.LogHelper.LogMessage("AutoShipStream: batching tick failed:" + Environment.NewLine + ex.ToString());
                   }

                   AutoShipStreamControl.Sleep(AutoShipStreamConfig.BatchIntervalMs);
               }
           }
           finally
           {
               ShipLink.Helpers.LogHelper.LogMessage("AutoShipStream: batching loop stopped.");
           }
       }

        /// <summary>
  /// Creates a batch and claims up to BatchSize Ready records into it, lowest Id
  /// first.
  ///
  /// Three statements in one transaction, written out here rather than hidden in
  /// a stored procedure so the whole claim can be stepped through and each step's
  /// result inspected.
  ///
  /// Two details are load-bearing:
  ///
  ///  - ORDER BY Id on the claim. Id is the arrival order, so taking the lowest
  ///    Ids first is what keeps batches themselves in order.
  ///  - No READPAST. Skipping a locked Ready row would push it into a later batch
  ///    and print it late - the exact reordering this design exists to prevent.
  ///    Block and wait instead.
  /// </summary>
  private static void ClaimBatch()
  {
      object pcId = DBNull.Value;
      if (Config.License != null && Config.License.CustomerPC != null)
      {
          pcId = Config.License.CustomerPC.Id;
      }

      using (SqlConnection connection = AutoShipDb.Open())
      using (SqlTransaction transaction = connection.BeginTransaction())
      {
          try
          {
              // 1. Create the batch so the claim below has something to point at.
              long batchId;
              using (SqlCommand command = AutoShipDb.Command(
                  connection,
                  @"INSERT INTO dbo.ShipmentBatch (Status, RecordCount, PCId, CreatedDate)
                    VALUES (@Status, 0, @PCId, GETUTCDATE());
                    SELECT CAST(SCOPE_IDENTITY() AS BIGINT);",
                  transaction))
              {
                  command.Parameters.Add(AutoShipDb.P("@Status", BatchStatus.Processing));
                  command.Parameters.Add(AutoShipDb.P("@PCId", pcId));

                  batchId = Convert.ToInt64(command.ExecuteScalar());
              }

              // 2. Claim the oldest Ready records into it.
              int claimed;
              using (SqlCommand command = AutoShipDb.Command(
                  connection,
                  @";WITH ready AS (
                        SELECT TOP (@BatchSize) Id, ShipStatus, ShipmentBatch_Id, BatchedAt
                          FROM dbo.ShipmentData WITH (UPDLOCK, ROWLOCK)
                         WHERE ShipStatus = @Ready
                         ORDER BY Id
                    )
                    UPDATE ready
                       SET ShipStatus       = @Processing,
                           ShipmentBatch_Id = @BatchId,
                           BatchedAt        = SYSUTCDATETIME();",
                  transaction))
              {
                  command.Parameters.Add(AutoShipDb.P("@BatchSize", AutoShipStreamConfig.BatchSize));
                  command.Parameters.Add(AutoShipDb.P("@Ready", ShipStatus.Ready));
                  command.Parameters.Add(AutoShipDb.P("@Processing", ShipStatus.Processing));
                  command.Parameters.Add(AutoShipDb.P("@BatchId", batchId));

                  claimed = command.ExecuteNonQuery();
              }

              // 3. Either seal the batch or undo it.
              if (claimed == 0)
              {
                  // Nothing was waiting. Don't leave an empty batch for the
                  // printing loop to pick up and immediately complete.
                  using (SqlCommand command = AutoShipDb.Command(
                      connection, "DELETE FROM dbo.ShipmentBatch WHERE Id = @Id;", transaction))
                  {
                      command.Parameters.Add(AutoShipDb.P("@Id", batchId));
                      command.ExecuteNonQuery();
                  }

                  transaction.Commit();
                  return;
              }

              using (SqlCommand command = AutoShipDb.Command(
                  connection,
                  "UPDATE dbo.ShipmentBatch SET RecordCount = @Count WHERE Id = @Id;",
                  transaction))
              {
                  command.Parameters.Add(AutoShipDb.P("@Count", claimed));
                  command.Parameters.Add(AutoShipDb.P("@Id", batchId));
                  command.ExecuteNonQuery();
              }

              transaction.Commit();

              ShipLink.Helpers.LogHelper.LogMessage("AutoShipStream: batch " + batchId + " created with " + claimed + " record(s).");
          }
          catch (Exception)
          {
              try
              {
                  transaction.Rollback();
              }
              catch (Exception) { }

              throw;
          }
      }
  }

  /// <summary>
  /// Fails every in-flight record that has outrun its deadline.
  ///
  /// This is what makes the whole design work. It guarantees each record reaches
  /// a terminal state in bounded time, so a batch always drains and the printing
  /// loop never needs to skip one - which is what would break print order.
  ///
  /// Two deadlines, because a record can be stuck in two different ways. The
  /// carrier deadline covers a call that was made and never answered, measured
  /// from ShipStartedAt. The unstarted deadline covers a record the shipping code
  /// never picked up at all, measured from BatchedAt - without it, a record with
  /// a null ShipStartedAt would sit in Processing forever and stall every label
  /// behind it.
  /// </summary>
  private static void SweepDeadlines()
  {
      const string Sql =
          @"UPDATE dbo.ShipmentData
               SET ShipStatus   = @Failed,
                   IsTimeOut    = 1,
                   ErrorMessage = @Error,
                   ShippedAt    = SYSUTCDATETIME()
             WHERE ShipStatus = @Processing
               AND (
                     (ShipStartedAt IS NOT NULL
                      AND ShipStartedAt < DATEADD(SECOND, -@CarrierSeconds, SYSUTCDATETIME()))
                  OR (ShipStartedAt IS NULL
                      AND BatchedAt IS NOT NULL
                      AND BatchedAt < DATEADD(SECOND, -@UnstartedSeconds, SYSUTCDATETIME()))
                   );";

      int expired = AutoShipDb.Execute(
          Sql,
          AutoShipDb.P("@Failed", ShipStatus.Failed),
          AutoShipDb.P("@Processing", ShipStatus.Processing),
          AutoShipDb.P("@Error", "Carrier did not respond within the configured deadline."),
          AutoShipDb.P("@CarrierSeconds", AutoShipStreamConfig.CarrierDeadlineSeconds),
          AutoShipDb.P("@UnstartedSeconds", AutoShipStreamConfig.UnstartedDeadlineSeconds));

      if (expired > 0)
      {
          // Worth a line every time. A steady trickle here means the carrier is
          // degraded and parcels are going out with error slips instead of labels,
          // which is not something to discover from the printer.
          ShipLink.Helpers.LogHelper.LogMessage(
              "AutoShipStream: " + expired + " record(s) exceeded the carrier deadline (" +
              AutoShipStreamConfig.CarrierDeadlineSeconds + "s) and were failed as timed out.");
      }
  }
       // -------------------------------------------------------------------------
       // Called by the carrier-facing code. Kept here so all writes to the shipping
       // columns go through one place.
       // -------------------------------------------------------------------------

       /// <summary>
       /// Starts the carrier deadline clock for a record about to be sent.
       /// Idempotent: only the first call stamps it, so a retry cannot extend a
       /// record's deadline indefinitely.
       /// </summary>
       public static void MarkShipStarted(long shipmentDataId)
       {
           AutoShipDb.Execute(
               @"UPDATE dbo.ShipmentData
                SET ShipStartedAt = SYSUTCDATETIME(),
                    AttemptCount  = AttemptCount + 1
              WHERE Id = @Id
                AND ShipStatus = @Processing
                AND ShipStartedAt IS NULL;",
               AutoShipDb.P("@Id", shipmentDataId),
               AutoShipDb.P("@Processing", ShipStatus.Processing));
       }

       /// <summary>
       /// Records a successful shipment. Guarded on the record still being in
       /// Processing, so a late carrier response cannot resurrect a record the
       /// deadline sweep already failed and whose NoLabel slip may have printed.
       /// </summary>
       /// <returns>False if the record had already moved on.</returns>
       public static bool MarkShipped(long shipmentDataId, string zpl, string trackingNumber, string carrierCode)
       {
           int affected = AutoShipDb.Execute(
               @"UPDATE dbo.ShipmentData
                SET ShipStatus     = @Shipped,
                    ZPL            = @Zpl,
                    TrackingNumber = @Tracking,
                    CarrierCode    = @Carrier,
                    ShippedAt      = SYSUTCDATETIME()
              WHERE Id = @Id
                AND ShipStatus = @Processing;",
               AutoShipDb.P("@Id", shipmentDataId),
               AutoShipDb.P("@Shipped", ShipStatus.Shipped),
               AutoShipDb.P("@Processing", ShipStatus.Processing),
               AutoShipDb.P("@Zpl", zpl),
               AutoShipDb.P("@Tracking", trackingNumber),
               AutoShipDb.P("@Carrier", carrierCode));

           return affected == 1;
       }

       /// <summary>Records a carrier refusal or error. Same guard as MarkShipped.</summary>
       public static bool MarkShipFailed(long shipmentDataId, string errorMessage, bool isTimeOut)
       {
           int affected = AutoShipDb.Execute(
               @"UPDATE dbo.ShipmentData
                SET ShipStatus   = @Failed,
                    IsTimeOut    = @IsTimeOut,
                    ErrorMessage = @Error,
                    ShippedAt    = SYSUTCDATETIME()
              WHERE Id = @Id
                AND ShipStatus = @Processing;",
               AutoShipDb.P("@Id", shipmentDataId),
               AutoShipDb.P("@Failed", ShipStatus.Failed),
               AutoShipDb.P("@Processing", ShipStatus.Processing),
               AutoShipDb.P("@IsTimeOut", isTimeOut),
               AutoShipDb.P("@Error", AutoShipDb.Truncate(errorMessage, 4000)));

           return affected == 1;
       }

       /// <summary>
       /// The records in a batch that are waiting on the carrier, in print order.
       /// Returns Id, OrderNumber and Data so the shipping code does not have to
       /// re-read the raw record.
       /// </summary>
       public static DataTable GetPendingCarrierWork(long batchId)
       {
           return AutoShipDb.Query(
               @"SELECT Id, OrderNumber, Data
               FROM dbo.ShipmentData
              WHERE ShipmentBatch_Id = @BatchId
                AND ShipStatus = @Processing
              ORDER BY PrintSeq;",
               AutoShipDb.P("@BatchId", batchId),
               AutoShipDb.P("@Processing", ShipStatus.Processing));
       }
   }   