You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: buffer_system_window.go
    path of file: ./repos/benthos/internal/impl/pure
    
    the code of the file until where you have to start completion: package pure

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/benthosdev/benthos/v4/internal/batch"
	"github.com/benthosdev/benthos/v4/internal/value"
	"github.com/benthosdev/benthos/v4/public/bloblang"
	"github.com/benthosdev/benthos/v4/public/service"
)

func tumblingWindowBufferConfig() *service.ConfigSpec {
	return service.NewConfigSpec().
		Beta().
		Version("3.53.0").
		Categories("Windowing").
		Summary("Chops a stream of messages into tumbling or sliding windows of fixed temporal size, following the system clock.").
		Description(`
A window is a grouping of messages that fit within a discrete measure of time following the system clock. Messages are allocated to a window either by the processing time (the time at which they're ingested) or by the event time, and this is controlled via the `+"[`timestamp_mapping` field](#timestamp_mapping)"+`.

In tumbling mode (default) the beginning of a window immediately follows the end of a prior window. When the buffer is initialized the first window to be created and populated is aligned against the zeroth minute of the zeroth hour of the day by default, and may therefore be open for a shorter period than the specified size.

A window is flushed only once the system clock surpasses i
    
func (w *systemWindowBuffer) WriteBatch(ctx context.Context, msgBatch service.MessageBatch, aFn service.AckFunc) error {
	w.pendingMut.Lock()
	defer w.pendingMut.Unlock()

	// If our output is blocked and therefore we haven't flushed more than the
	// last two windows we purge messages that wouldn't fit within them.
	prevStart, _, _, _ := w.nextSystemWindow()
	if w.latestFlushedWindowEnd.Before(prevStart) && w.oldestTS.Before(prevStart) {
		newOldestTS := w.clock()
		newPending := make([]*tsMessage, 0, len(w.pending))
		for _, pending := range w.pending {
			if pending.ts.Before(prevStart) {
				// Reject messages too old to fit into a window by acknowledging
				// them.
				_ = pending.ackFn(ctx, nil)
				continue
			}
			newPending = append(newPending, pending)
			if pending.ts.Before(newOldestTS) {
				newOldestTS = pending.ts
			}
		}
		w.pending = newPending
	}

	messageAdded := false
	aggregatedAck := batch.NewCombinedAcker(batch.AckFunc(aFn))

	// And now add new messages.
	for i, msg := range msgBatch {
		ts, err := w.getTimestamp(i, msgBatch)
		if err != nil {
			return err
		}

		// Don't add messages older than our current window start.
		if !ts.After(w.latestFlushedWindowEnd) { //nolint: gocritic
			continue
		}

		messageAdded = true
		w.pending = append(w.pending, &tsMessage{
			ts: ts, m: msg, ackFn: service.AckFunc(aggregatedAck.Derive()),
		})
		if ts.Before(w.oldestTS) {
			w.oldestTS = ts
		}
	}

	if !messageAdded {
		// If none of the messages have fit into a window we reject them by
		// acknowledging the batch.
		_ = aFn(ctx, nil)
	}
	return nil
}