streamplace / pkg / spxrpc / webhook.go

package spxrpc

import (
	"context"
	"encoding/json"
	"net/http"
	"net/url"
	"strconv"
	"strings"

	"github.com/labstack/echo/v4"
	"github.com/streamplace/oatproxy/pkg/oatproxy"
	"stream.place/streamplace/pkg/log"
	"stream.place/streamplace/pkg/statedb"
	placestreamtypes "stream.place/streamplace/pkg/streamplace"
)

func (s *Server) handlePlaceStreamServerCreateWebhook(ctx context.Context, input *placestreamtypes.ServerCreateWebhook_Input) (*placestreamtypes.ServerCreateWebhook_Output, error) {
	// Get authenticated user
	session, _ := oatproxy.GetOAuthSession(ctx)
	if session == nil {
		return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found")
	}

	// Validate input
	if input.Url == "" {
		return nil, echo.NewHTTPError(http.StatusBadRequest, "URL is required")
	}
	if len(input.Events) == 0 {
		return nil, echo.NewHTTPError(http.StatusBadRequest, "At least one event type is required")
	}

	// Validate URL format
	if _, err := url.Parse(input.Url); err != nil {
		return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid URL format")
	}
	// Convert input to database model using the conversion function
	webhook, err := statedb.WebhookFromLexiconInput(input, session.DID, "") // ID will be generated by the database
	if err != nil {
		log.Error(ctx, "failed to convert input to webhook", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to create webhook")
	}

	// Create webhook
	err = s.statefulDB.CreateWebhook(webhook)
	if err != nil {
		log.Error(ctx, "failed to create webhook", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to create webhook")
	}

	// Convert to API response
	apiWebhook, err := webhook.ToLexicon()
	if err != nil {
		log.Error(ctx, "failed to convert webhook to API format", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to format webhook response")
	}

	return &placestreamtypes.ServerCreateWebhook_Output{
		Webhook: apiWebhook,
	}, nil
}

func (s *Server) handlePlaceStreamServerListWebhooks(ctx context.Context, active *bool, cursor string, event string, limit int) (*placestreamtypes.ServerListWebhooks_Output, error) {
	// Get authenticated user
	session, _ := oatproxy.GetOAuthSession(ctx)
	if session == nil {
		return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found")
	}

	// Set default limit
	if limit <= 0 || limit > 100 {
		limit = 50
	}

	// Parse cursor for offset
	offset := 0
	if cursor != "" {
		var err error
		offset, err = strconv.Atoi(cursor)
		if err != nil {
			return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid cursor")
		}
	}

	// Build filters
	filters := make(map[string]interface{})
	if active != nil {
		filters["active"] = *active
	}

	// Get webhooks
	webhooks, err := s.statefulDB.ListWebhooks(session.DID, limit+1, offset, filters)
	if err != nil {
		log.Error(ctx, "failed to list webhooks", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to list webhooks")
	}

	// Filter by event type if specified
	if event != "" {
		filtered := make([]statedb.Webhook, 0)
		for _, w := range webhooks {
			var events []string
			if err := json.Unmarshal(w.Events, &events); err == nil {
				for _, e := range events {
					if e == event {
						filtered = append(filtered, w)
						break
					}
				}
			}
		}
		webhooks = filtered
	}

	// Check if there are more results
	var nextCursor *string
	if len(webhooks) > limit {
		webhooks = webhooks[:limit]
		next := strconv.Itoa(offset + limit)
		nextCursor = &next
	}

	// Convert to API format
	apiWebhooks := make([]*placestreamtypes.ServerDefs_Webhook, len(webhooks))
	for i, webhook := range webhooks {
		apiWebhook, err := webhook.ToLexicon()
		if err != nil {
			log.Error(ctx, "failed to convert webhook to API format", "err", err)
			continue
		}
		apiWebhooks[i] = apiWebhook
	}

	return &placestreamtypes.ServerListWebhooks_Output{
		Webhooks: apiWebhooks,
		Cursor:   nextCursor,
	}, nil
}

func (s *Server) handlePlaceStreamServerGetWebhook(ctx context.Context, id string) (*placestreamtypes.ServerGetWebhook_Output, error) {
	// Get authenticated user
	session, _ := oatproxy.GetOAuthSession(ctx)
	if session == nil {
		return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found")
	}

	// Get webhook
	webhook, err := s.statefulDB.GetWebhook(id, session.DID)
	if err != nil {
		if strings.Contains(err.Error(), "record not found") {
			return nil, echo.NewHTTPError(http.StatusNotFound, "Webhook not found")
		}
		log.Error(ctx, "failed to get webhook", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to get webhook")
	}

	// Convert to API format
	apiWebhook, err := webhook.ToLexicon()
	if err != nil {
		log.Error(ctx, "failed to convert webhook to API format", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to format webhook response")
	}

	return &placestreamtypes.ServerGetWebhook_Output{
		Webhook: apiWebhook,
	}, nil
}

func (s *Server) handlePlaceStreamServerUpdateWebhook(ctx context.Context, input *placestreamtypes.ServerUpdateWebhook_Input) (*placestreamtypes.ServerUpdateWebhook_Output, error) {
	// Get authenticated user
	session, _ := oatproxy.GetOAuthSession(ctx)
	if session == nil {
		return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found")
	}

	// Validate URL if provided
	if input.Url != nil {
		if _, err := url.Parse(*input.Url); err != nil {
			return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid URL format")
		}
	}

	// Build updates map
	updates := make(map[string]interface{})
	if input.Url != nil {
		updates["url"] = *input.Url
	}
	if input.Events != nil {
		eventsJSON, err := json.Marshal(input.Events)
		if err != nil {
			return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid events format")
		}
		updates["events"] = json.RawMessage(eventsJSON)
	}
	if input.Active != nil {
		updates["active"] = *input.Active
	}
	if input.Prefix != nil {
		updates["prefix"] = *input.Prefix
	}
	if input.Suffix != nil {
		updates["suffix"] = *input.Suffix
	}
	if input.Rewrite != nil {
		rewriteJSON, err := json.Marshal(input.Rewrite)
		if err != nil {
			return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid rewrite rules format")
		}
		updates["rewrite"] = json.RawMessage(rewriteJSON)
	}
	if input.Name != nil {
		updates["name"] = *input.Name
	}
	if input.Description != nil {
		updates["description"] = *input.Description
	}
	if input.MuteWords != nil {
		muteWordsJSON, err := json.Marshal(input.MuteWords)
		if err != nil {
			return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid mute words format")
		}
		updates["mute_words"] = json.RawMessage(muteWordsJSON)
	}

	if len(updates) == 0 {
		return nil, echo.NewHTTPError(http.StatusBadRequest, "No fields to update")
	}

	// Update webhook
	webhook, err := s.statefulDB.UpdateWebhook(input.Id, session.DID, updates)
	if err != nil {
		if strings.Contains(err.Error(), "record not found") {
			return nil, echo.NewHTTPError(http.StatusNotFound, "Webhook not found")
		}
		log.Error(ctx, "failed to update webhook", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to update webhook")
	}

	// Convert to API format
	apiWebhook, err := webhook.ToLexicon()
	if err != nil {
		log.Error(ctx, "failed to convert webhook to API format", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to format webhook response")
	}

	return &placestreamtypes.ServerUpdateWebhook_Output{
		Webhook: apiWebhook,
	}, nil
}

func (s *Server) handlePlaceStreamServerDeleteWebhook(ctx context.Context, input *placestreamtypes.ServerDeleteWebhook_Input) (*placestreamtypes.ServerDeleteWebhook_Output, error) {
	// Get authenticated user
	session, _ := oatproxy.GetOAuthSession(ctx)
	if session == nil {
		return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found")
	}

	// Delete webhook
	err := s.statefulDB.DeleteWebhook(input.Id, session.DID)
	if err != nil {
		log.Error(ctx, "failed to delete webhook", "err", err)
		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to delete webhook")
	}

	return &placestreamtypes.ServerDeleteWebhook_Output{
		Success: true,
	}, nil
}