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: event_time_buffer.go
    path of file: ./repos/octosql/execution/nodes
    
    the code of the file until where you have to start completion: package nodes

import (
	"fmt"

	. "github.com/cube2222/octosql/execution"
)

type EventTimeBuffer struct {
	source Node
}

func NewEventTimeBuffer(source Node) *EventTimeB
    
func (e *EventTimeBuffer) Run(ctx ExecutionContext, produce ProduceFn, metaSend MetaSendFn) error {
	records := NewRecordEventTimeBuffer()

	if err := e.source.Run(
		ctx,
		func(ctx ProduceContext, record Record) error {
			if record.EventTime.IsZero() {
				// If the event time is zero, don't buffer, there's no point.
				// There won't be any record with an event time less than zero.
				return produce(ctx, record)
			}
			records.AddRecord(record)
			return nil
		},
		func(ctx ProduceContext, msg MetadataMessage) error {
			if msg.Type == MetadataMessageTypeWatermark {
				if err := records.Emit(msg.Watermark, ProduceFnApplyContext(produce, ctx)); err != nil {
					return fmt.Errorf("couldn't emit records up to watermark: %w", err)
				}
			}
			return metaSend(ctx, msg)
		}); err != nil {
		return err
	}

	if err := records.Emit(WatermarkMaxValue, ProduceFnApplyContext(produce, ProduceFromExecutionContext(ctx))); err != nil {
		return fmt.Errorf("couldn't emit remaining records: %w", err)
	}
	return nil
}