fn_registry/functions/infra/matrix/matrix_sync_service_test.go

//go:build goolm

package matrix

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"

	"maunium.net/go/mautrix"
	"maunium.net/go/mautrix/id"
)

// fakeSynapseServer crea un httptest.Server que simula Synapse para tests de sync.
// syncHandler recibe el numero de llamada /sync (1-indexed) y devuelve la respuesta.
// nil response significa bloquear hasta ctx cancelado.
func fakeSynapseServer(t *testing.T, syncFn func(call int, w http.ResponseWriter, r *http.Request)) *httptest.Server {
	t.Helper()
	var callCount int32
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		switch {
		case r.Method == http.MethodPost && r.URL.Path == "/_matrix/client/v3/user/@alice:localhost/filter":
			// mautrix necesita este endpoint para guardar el filtro; responder con un filter_id
			_ = json.NewEncoder(w).Encode(map[string]interface{}{"filter_id": "f1"})
		case r.URL.Path == "/_matrix/client/v3/sync" || r.URL.Path == "/_matrix/client/r0/sync":
			n := int(atomic.AddInt32(&callCount, 1))
			syncFn(n, w, r)
		default:
			w.WriteHeader(http.StatusNotFound)
		}
	}))
}

// syncRespMessage construye una respuesta /sync con un m.room.message.
func syncRespMessage(nextBatch string) map[string]interface{} {
	return map[string]interface{}{
		"next_batch": nextBatch,
		"rooms": map[string]interface{}{
			"join": map[string]interface{}{
				"!testroom:localhost": map[string]interface{}{
					"timeline": map[string]interface{}{
						"events": []interface{}{
							map[string]interface{}{
								"event_id":         "$evt001:localhost",
								"type":             "m.room.message",
								"sender":           "@alice:localhost",
								"origin_server_ts": int64(1700000000000),
								"room_id":          "!testroom:localhost",
								"content": map[string]interface{}{
									"msgtype": "m.text",
									"body":    "hola mundo",
								},
							},
						},
						"limited": false,
					},
				},
			},
		},
	}
}

// newTestSyncClient crea un *mautrix.Client apuntando al servidor dado.
func newTestSyncClient(t *testing.T, serverURL string) *mautrix.Client {
	t.Helper()
	cli, err := mautrix.NewClient(serverURL, "@alice:localhost", "token-test")
	if err != nil {
		t.Fatalf("NewClient: %v", err)
	}
	cli.UserID = id.UserID("@alice:localhost")
	return cli
}

// TestMatrixSyncService_RecibeMensajeYStop arranca el servicio, recibe 1 evento y hace Stop limpio.
func TestMatrixSyncService_RecibeMensajeYStop(t *testing.T) {
	srv := fakeSynapseServer(t, func(n int, w http.ResponseWriter, r *http.Request) {
		if n == 1 {
			_ = json.NewEncoder(w).Encode(syncRespMessage("nb_001"))
			return
		}
		// Bloquear syncs subsiguientes hasta cancelacion
		<-r.Context().Done()
		_ = json.NewEncoder(w).Encode(map[string]interface{}{"next_batch": "nb_002"})
	})
	defer srv.Close()

	cli := newTestSyncClient(t, srv.URL)
	h, err := MatrixSyncService(context.Background(), MatrixSyncServiceConfig{
		Client:        cli,
		ChannelBuffer: 16,
	})
	if err != nil {
		t.Fatalf("MatrixSyncService: %v", err)
	}

	// Esperar el primer evento
	select {
	case ev, ok := <-h.Events:
		if !ok {
			t.Fatal("canal cerrado antes de recibir evento")
		}
		if ev.Type != "message" {
			t.Errorf("tipo esperado 'message', got %q", ev.Type)
		}
		if ev.Body != "hola mundo" {
			t.Errorf("body esperado 'hola mundo', got %q", ev.Body)
		}
		if ev.Sender != "@alice:localhost" {
			t.Errorf("sender esperado '@alice:localhost', got %q", ev.Sender)
		}
		if ev.RoomID != "!testroom:localhost" {
			t.Errorf("roomID esperado '!testroom:localhost', got %q", ev.RoomID)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("timeout esperando evento")
	}

	// Stop limpio
	h.Stop()

	// Verificar que Events cierra tras Stop
	timeout := time.After(3 * time.Second)
	for {
		select {
		case _, ok := <-h.Events:
			if !ok {
				return // canal cerrado correctamente
			}
		case <-timeout:
			t.Fatal("canal Events no cerro tras Stop")
		}
	}
}

// TestMatrixSyncService_BackoffRecovery verifica backoff con 2 errores 500 seguidos de exito.
func TestMatrixSyncService_BackoffRecovery(t *testing.T) {
	srv := fakeSynapseServer(t, func(n int, w http.ResponseWriter, r *http.Request) {
		if n <= 2 {
			// Primeras 2 llamadas: devolver 500
			w.WriteHeader(http.StatusInternalServerError)
			_ = json.NewEncoder(w).Encode(map[string]interface{}{
				"errcode": "M_UNKNOWN",
				"error":   "internal server error",
			})
			return
		}
		if n == 3 {
			// Tercera llamada: respuesta correcta inmediata (no bloquear)
			_ = json.NewEncoder(w).Encode(syncRespMessage("nb_recovery"))
			return
		}
		// Cuarta en adelante: bloquear hasta cancelacion
		<-r.Context().Done()
		_ = json.NewEncoder(w).Encode(map[string]interface{}{"next_batch": "nb_x"})
	})
	defer srv.Close()

	cli := newTestSyncClient(t, srv.URL)
	h, err := MatrixSyncService(context.Background(), MatrixSyncServiceConfig{
		Client:           cli,
		InitialBackoffMS: 50, // backoff corto para tests
		MaxBackoffMS:     200,
		ChannelBuffer:    16,
	})
	if err != nil {
		t.Fatalf("MatrixSyncService: %v", err)
	}
	defer h.Stop()

	// Tras los fallos, debe llegar el evento de recovery
	select {
	case ev, ok := <-h.Events:
		if !ok {
			t.Fatal("canal cerrado antes de evento de recovery")
		}
		if ev.Type != "message" {
			t.Errorf("tipo esperado 'message', got %q", ev.Type)
		}
		if ev.Body != "hola mundo" {
			t.Errorf("body esperado 'hola mundo', got %q", ev.Body)
		}
	case <-time.After(8 * time.Second):
		t.Fatal("timeout esperando evento de recovery tras backoff")
	}
}

// TestMatrixSyncService_Error401NoExit verifica que 401 emite error pero no cierra el servicio.
func TestMatrixSyncService_Error401NoExit(t *testing.T) {
	srv := fakeSynapseServer(t, func(n int, w http.ResponseWriter, r *http.Request) {
		if n == 1 {
			// Primera llamada: 401 M_UNKNOWN_TOKEN
			w.WriteHeader(http.StatusUnauthorized)
			_ = json.NewEncoder(w).Encode(map[string]interface{}{
				"errcode": "M_UNKNOWN_TOKEN",
				"error":   "Invalid macaroon passed.",
			})
			return
		}
		// Bloquear: el servicio espera en backoff maximo
		<-r.Context().Done()
		_ = json.NewEncoder(w).Encode(map[string]interface{}{"next_batch": "nb_x"})
	})
	defer srv.Close()

	cli := newTestSyncClient(t, srv.URL)
	h, err := MatrixSyncService(context.Background(), MatrixSyncServiceConfig{
		Client:           cli,
		InitialBackoffMS: 50,
		MaxBackoffMS:     200,
		ChannelBuffer:    8,
	})
	if err != nil {
		t.Fatalf("MatrixSyncService: %v", err)
	}

	// Debe recibir al menos un error (fatal 401)
	select {
	case syncErr := <-h.Errors:
		if syncErr == nil {
			t.Error("error esperado no nil")
		}
	case <-time.After(4 * time.Second):
		t.Fatal("timeout esperando error 401 en canal Errors")
	}

	// El canal Events NO debe estar cerrado — el servicio sigue activo
	select {
	case _, ok := <-h.Events:
		if !ok {
			t.Fatal("canal Events no debia cerrarse con error 401 (dejar al caller decidir via Stop)")
		}
	case <-time.After(300 * time.Millisecond):
		// Correcto: canal sigue abierto
	}

	h.Stop()

	// Tras Stop, Events debe cerrarse
	select {
	case _, ok := <-h.Events:
		if !ok {
			return // OK
		}
	case <-time.After(3 * time.Second):
		t.Fatal("canal Events no cerro tras Stop despues de error 401")
	}
}

// TestMatrixSyncService_StopIdempotente verifica que Stop() dos veces no causa panic.
func TestMatrixSyncService_StopIdempotente(t *testing.T) {
	srv := fakeSynapseServer(t, func(n int, w http.ResponseWriter, r *http.Request) {
		<-r.Context().Done()
		_ = json.NewEncoder(w).Encode(map[string]interface{}{"next_batch": "nb_1"})
	})
	defer srv.Close()

	cli := newTestSyncClient(t, srv.URL)
	h, err := MatrixSyncService(context.Background(), MatrixSyncServiceConfig{
		Client: cli,
	})
	if err != nil {
		t.Fatalf("MatrixSyncService: %v", err)
	}

	// Llamar Stop dos veces — no debe panic
	defer func() {
		if r := recover(); r != nil {
			t.Errorf("Stop() dos veces causó panic: %v", r)
		}
	}()
	h.Stop()
	h.Stop()
}

// TestMatrixSyncService_CtxCancelCierraChannels verifica que cancelar ctx cierra Events < 1s.
func TestMatrixSyncService_CtxCancelCierraChannels(t *testing.T) {
	srv := fakeSynapseServer(t, func(n int, w http.ResponseWriter, r *http.Request) {
		<-r.Context().Done()
		_ = json.NewEncoder(w).Encode(map[string]interface{}{"next_batch": "nb_ctx"})
	})
	defer srv.Close()

	ctx, cancel := context.WithCancel(context.Background())
	cli := newTestSyncClient(t, srv.URL)
	h, err := MatrixSyncService(ctx, MatrixSyncServiceConfig{
		Client:        cli,
		ChannelBuffer: 4,
	})
	if err != nil {
		t.Fatalf("MatrixSyncService: %v", err)
	}

	// Cancelar contexto padre
	cancel()

	// Events debe cerrarse en menos de 1 segundo
	deadline := time.After(1 * time.Second)
	for {
		select {
		case _, ok := <-h.Events:
			if !ok {
				return // canal cerrado correctamente
			}
		case <-deadline:
			t.Fatal("canal Events no cerro en 1s tras cancelar ctx")
		}
	}
}