nodb/internal/server/server.go

package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"strings"

	databasev1 "github.com/iwanhae/nodb/api/v1"
	"github.com/iwanhae/nodb/pkg/broadcaster"
	"github.com/labstack/echo/v4"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type Server struct {
	client.Client
	Broadcaster broadcaster.Broadcaster[runtime.Object]
}

func (s *Server) delete(c echo.Context) error {
	ctx := c.Request().Context()
	dbType := c.Param("type")
	namespace := c.Param("namespace")
	name := c.Param("name")

	switch strings.ToLower(dbType) {
	case "postgresql":
		return s.Client.Delete(ctx, &databasev1.PostgreSQL{
			ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}},
		)
	default:
		return echo.NewHTTPError(http.StatusBadRequest)
	}
}

func (s *Server) create(c echo.Context) error {
	ctx := c.Request().Context()
	dbType := c.Param("type")
	switch strings.ToLower(dbType) {
	case "postgresql":
		obj := &databasev1.PostgreSQL{}
		if err := json.NewDecoder(c.Request().Body).Decode(obj); err != nil {
			return err
		}
		return s.Client.Create(ctx, obj)
	default:
		return echo.NewHTTPError(http.StatusBadRequest)
	}
}

func (s *Server) watchUpdates(c echo.Context) error {
	ctx := c.Request().Context()
	ch := s.Broadcaster.Subscribe()

	w := c.Response().Writer
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Access-Control-Allow-Origin", "*")

	flusher := w.(http.Flusher)
	{
		list := databasev1.PostgreSQLList{}
		if err := s.Client.List(ctx, &list); err != nil {
			return err
		}
		go func() {
			for _, item := range list.Items {
				ch <- item.DeepCopy()
			}
		}()
	}

	go func() {
		<-c.Request().Context().Done()
		close(ch)
	}()

	for obj := range ch {
		fmt.Fprint(w, "data: ")
		json.NewEncoder(w).Encode(obj)
		fmt.Fprint(w, "\n\n")
		flusher.Flush()
	}

	return nil
}