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: kv.go
    path of file: ./repos/streamdal/apps/server/apis/httpapi
    
    the code of the file until where you have to start completion: package httpapi

import (
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"

	"github.com/julienschmidt/httprouter"
	"google.golang.org/protobuf/encoding/protojson"

	"github.com/streamdal/streamdal/libs/protos/build/go/protos"

	"github.com/streamdal/streamdal/apps/server/validate"
)

func (a *HTTPAPI) getUsageKVHandler(rw http.ResponseWriter, r *http.Request) {
	rw.Header().Set("Content-Type", "application/json; charset=UTF-8")

	usage, err := a.Options.KVService.GetUsage(r.Context())
	if err != nil {
		Write(rw, http.StatusInternalServerError, "unable to fetch kv usage: %v", err)
		return
	}

	WriteJSON(rw, usage, http.StatusOK)
}

// Returns []*protos.KVObject as JSON
func (a *HTTPAPI) getAllKVHandler(rw http.ResponseWriter, r *http.Request) {
	rw.Header().Set("Content-Type", "application/json; charset=UTF-8")

	kvs, err := a.Options.KVService.GetAll(r.Context())
	if err != nil {
		Write(rw, http.StatusInternalServerError, "unable to fetch all kv objects: %v", err)
		return
	}

	WriteJSON(rw, kvs, http.StatusOK)
}

// Returns *protos.KVObject as JSON
func (a *H
    
func (a *HTTPAPI) createKVHandler(rw http.ResponseWriter, r *http.Request) {
	rw.Header().Set("Content-Type", "application/json; charset=UTF-8")

	body, err := io.ReadAll(r.Body)
	if err != nil {
		Write(rw, http.StatusBadRequest, "unable to read request body: %v", err)
		return
	}

	defer r.Body.Close()

	// Unmarshal input into pb's
	createHTTPRequest := &protos.KVCreateHTTPRequest{}

	if err := protojson.Unmarshal(body, createHTTPRequest); err != nil {
		Write(rw, http.StatusBadRequest, "unable to unmarshal request body: %v", err)
		return
	}

	// Validate the input
	if err := validate.KVCreateHTTPRequest(createHTTPRequest); err != nil {
		Write(rw, http.StatusBadRequest, "invalid request body: %v", err)
		return
	}

	// Attempt to create in KV service
	//
	// TODO: This should be updated to pass this data to the KV service via a chan;
	// this should be good enough for now though.
	if err := a.Options.KVService.Create(r.Context(), createHTTPRequest.Kvs, createHTTPRequest.Overwrite); err != nil {
		Write(rw, http.StatusInternalServerError, "unable to complete create KV request: %v", err)
		return
	}

	// Broadcast KV changes to other server nodes (reminder: we do this
	// so that the nodes can inform their connected SDKs of the change and the
	// SDKs can update their local KV state)
	// TODO: create broadcast handlers for emitting commands on KV updates
	if err := a.Options.BusService.BroadcastKVCreate(r.Context(), createHTTPRequest.Kvs, createHTTPRequest.Overwrite); err != nil {
		Write(rw, http.StatusInternalServerError, "unable to broadcast kv command: %v", err)
		return
	}

	WriteJSON(rw, createHTTPRequest.Kvs, http.StatusOK)
}