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: processor.go
    path of file: ./repos/benthos/internal/impl/mongodb
    
    the code of the file until where you have to start completion: package mongodb

import (
	"context"
	"errors"
	"fmt"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"

	"github.com/benthosdev/benthos/v4/internal/impl/pure"
	"github.com/benthosdev/benthos/v4/public/service"
)

const (
	mpFieldCollection      = "collection"
	mpFieldWriteConcern    = "write_concern"
	mpFieldJSONMarshalMode = "json_marshal_mode"
)

// ProcessorSpec defines
    
func (m *Processor) ProcessBatch(ctx context.Context, batch service.MessageBatch) ([]service.MessageBatch, error) {
	writeModelsMap := map[string]msgsAndModels{}

	_ = batch.WalkWithBatchedErrors(func(i int, msg *service.Message) (err error) {
		defer func() {
			if err != nil {
				msg.SetError(err)
			}
		}()

		docJSON, filterJSON, hintJSON, err := m.writeMaps.extractFromMessage(m.operation, i, batch)
		if err != nil {
			return err
		}

		findOptions := &options.FindOneOptions{}
		if hintJSON != nil {
			findOptions.Hint = hintJSON
		}

		collectionStr, err := batch.TryInterpolatedString(i, m.collection)
		if err != nil {
			return fmt.Errorf("collection interpolation error: %w", err)
		}

		var writeModel mongo.WriteModel
		switch m.operation {
		case OperationInsertOne:
			writeModel = &mongo.InsertOneModel{
				Document: docJSON,
			}
		case OperationDeleteOne:
			writeModel = &mongo.DeleteOneModel{
				Filter: filterJSON,
				Hint:   hintJSON,
			}
		case OperationDeleteMany:
			writeModel = &mongo.DeleteManyModel{
				Filter: filterJSON,
				Hint:   hintJSON,
			}
		case OperationReplaceOne:
			writeModel = &mongo.ReplaceOneModel{
				Upsert:      &m.writeMaps.upsert,
				Filter:      filterJSON,
				Replacement: docJSON,
				Hint:        hintJSON,
			}
		case OperationUpdateOne:
			writeModel = &mongo.UpdateOneModel{
				Upsert: &m.writeMaps.upsert,
				Filter: filterJSON,
				Update: docJSON,
				Hint:   hintJSON,
			}
		case OperationFindOne:
			collection := m.database.Collection(collectionStr, m.writeConcernCollectionOption)

			var decoded any
			if err = collection.FindOne(context.Background(), filterJSON, findOptions).Decode(&decoded); err != nil {
				if errors.Is(err, mongo.ErrNoDocuments) {
					return err
				}
				m.log.Errorf("Error decoding mongo db result, filter = %v: %s", filterJSON, err)
				return err
			}

			data, err := bson.MarshalExtJSON(decoded, m.marshalMode == JSONMarshalModeCanonical, false)
			if err != nil {
				return err
			}

			msg.SetBytes(data)
			return nil
		}

		if writeModel != nil {
			tmp := writeModelsMap[collectionStr]
			tmp.ws = append(tmp.ws, writeModel)
			tmp.msgs = append(tmp.msgs, msg)
			writeModelsMap[collectionStr] = tmp
		}
		return nil
	})

	if len(writeModelsMap) > 0 {
		for collectionStr, msAndMs := range writeModelsMap {
			collection := m.database.Collection(collectionStr, m.writeConcernCollectionOption)

			// We should have at least one write model in the slice
			if _, err := collection.BulkWrite(ctx, msAndMs.ws); err != nil {
				m.log.Errorf("Bulk write failed in mongodb processor: %v", err)
				for _, msg := range msAndMs.msgs {
					msg.SetError(err)
				}
			}
		}
	}

	return []service.MessageBatch{batch}, nil
}