logos-messaging-go-bindings/internal/ffi/liblogosdelivery.go

// Package ffi is the cgo bridge over the single liblogosdelivery C library,
// which exposes the full logos-delivery API. It is split across two files
// mirroring the two API tiers: this file (liblogosdelivery.go) owns the shared
// plumbing — the three reply-callback shapes, the pending-call registry and the
// node lifecycle — plus the stable Messaging API; libwaku.go adds the low-level
// Kernel API (waku_*). It exposes Go-typed primitives so pkg/kernel and
// pkg/messaging stay pure Go.
//
// The C surface is generated by nim-ffi from the {.ffi.} annotations in
// logos-delivery's library/*.nim into library/generated/logosdelivery.h, which
// library/liblogosdelivery.h includes. Every entry point takes the context
// handle first (except the constructor) and comes in one of two shapes:
//
//   - scalar fast path (no arguments, string return): a raw
//     LogosDeliveryScalarRawFn callback receives length-delimited bytes;
//   - everything else: a per-call LogosDelivery<Name>ReplyFn callback receives
//     a NUL-terminated reply or error, and the arguments ride in a per-call
//     <Name>Req struct passed last.
//
// All callback strings are borrowed for the duration of the call, so every
// callback here copies before it hands anything back to Go.
package ffi

/*
#cgo LDFLAGS: -llogosdelivery
#include <liblogosdelivery.h>
#include <stdint.h>
#include <stdlib.h>

// The three callback shapes the generated ABI uses. All are implemented in Go
// and exported below. `userData` carries a runtime/cgo.Handle cast to void*.
extern void logosScalarReply(int callerRet, char* msg, size_t len, void* userData);
extern void logosReply(int errCode, char* reply, char* errMsg, void* userData);
extern void logosCreated(int errCode, char* ctxAddr, char* errMsg, void* userData);
extern void logosEvent(int callerRet, char* msg, size_t len, void* userData);

// Thin wrappers binding the shared Go callbacks to each entry point. They take
// the cgo.Handle as a uintptr_t and widen it to void* here, so the Go side
// never converts a uintptr back into an unsafe.Pointer.
static void* cGoCreateNode(const char* configJson, uintptr_t ud) {
	LogosdeliveryCreateNodeCtorReq req;
	req.configJson = configJson;
	return logosdelivery_create_node(&req, (LogosDeliveryCreateRawFn) logosCreated, (void*) ud);
}
static int cGoStartNode(void* ctx, uintptr_t ud) {
	return logosdelivery_start_node(ctx, (LogosDeliveryScalarRawFn) logosScalarReply, (void*) ud);
}
static int cGoStopNode(void* ctx, uintptr_t ud) {
	return logosdelivery_stop_node(ctx, (LogosDeliveryScalarRawFn) logosScalarReply, (void*) ud);
}
static int cGoSubscribe(void* ctx, const char* contentTopic, uintptr_t ud) {
	LogosdeliverySubscribeReq req;
	req.contentTopicStr = contentTopic;
	return logosdelivery_subscribe(ctx, (LogosDeliverySubscribeReplyFn) logosReply, (void*) ud, &req);
}
static int cGoUnsubscribe(void* ctx, const char* contentTopic, uintptr_t ud) {
	LogosdeliveryUnsubscribeReq req;
	req.contentTopicStr = contentTopic;
	return logosdelivery_unsubscribe(ctx, (LogosDeliveryUnsubscribeReplyFn) logosReply, (void*) ud, &req);
}
static int cGoSend(void* ctx, const char* messageJson, uintptr_t ud) {
	LogosdeliverySendReq req;
	req.messageJson = messageJson;
	return logosdelivery_send(ctx, (LogosDeliverySendReplyFn) logosReply, (void*) ud, &req);
}
static uint64_t cGoAddEventListener(void* ctx, const char* eventName, uintptr_t ud) {
	return logosdelivery_add_event_listener(ctx, eventName, (FFICallBack) logosEvent, (void*) ud);
}
*/
import "C"

import (
	"errors"
	"fmt"
	"runtime/cgo"
	"sync"
	"unsafe"
)

// Handle is an opaque pointer to a node context owned by the C library.
type Handle = unsafe.Pointer

// RetOK is the return code callbacks report on success.
const RetOK = C.NIMFFI_RET_OK

// staleWarn is the non-terminal "still running" code the library emits every
// few seconds for a long call. It is always followed by a terminal code, so
// every callback below ignores it rather than settling the pending call.
const staleWarn = C.NIMFFI_RET_STALE_WARN

// ListenerID identifies one registered event listener within a node context.
// It is only meaningful together with the Handle it was registered on.
type ListenerID uint64

// EventHandler receives every event liblogosdelivery emits for the event name
// it was registered under: the raw event JSON when ret == RetOK, an error
// message otherwise.
type EventHandler func(ret int, msg string)

// pending is one in-flight synchronous call. The C callback settles it and the
// caller reads msg/err after done is closed.
type pending struct {
	done chan struct{}
	once sync.Once
	msg  string
	err  error
}

func newPending() *pending { return &pending{done: make(chan struct{})} }

// settle records a terminal result exactly once and wakes the caller.
func (p *pending) settle(ok bool, msg string) {
	p.once.Do(func() {
		if ok {
			p.msg = msg
		} else {
			if msg == "" {
				msg = "liblogosdelivery call failed"
			}
			p.err = errors.New(msg)
		}
		close(p.done)
	})
}

// goStringN copies a length-delimited, possibly non-NUL-terminated byte run.
func goStringN(s *C.char, length C.size_t) string {
	if s == nil || length == 0 {
		return ""
	}
	return C.GoStringN(s, C.int(length))
}

// goString copies a borrowed NUL-terminated string, tolerating NULL.
func goString(s *C.char) string {
	if s == nil {
		return ""
	}
	return C.GoString(s)
}

//export logosScalarReply
func logosScalarReply(callerRet C.int, msg *C.char, length C.size_t, userData unsafe.Pointer) {
	if callerRet == staleWarn {
		return
	}
	p, ok := pendingFrom(userData)
	if !ok {
		return
	}
	p.settle(callerRet == RetOK, goStringN(msg, length))
}

//export logosReply
func logosReply(errCode C.int, reply *C.char, errMsg *C.char, userData unsafe.Pointer) {
	if errCode == staleWarn {
		return
	}
	p, ok := pendingFrom(userData)
	if !ok {
		return
	}
	if errCode == RetOK {
		p.settle(true, goString(reply))
		return
	}
	p.settle(false, goString(errMsg))
}

//export logosCreated
func logosCreated(errCode C.int, ctxAddr *C.char, errMsg *C.char, userData unsafe.Pointer) {
	if errCode == staleWarn {
		return
	}
	p, ok := pendingFrom(userData)
	if !ok {
		return
	}
	if errCode == RetOK {
		p.settle(true, goString(ctxAddr))
		return
	}
	p.settle(false, goString(errMsg))
}

//export logosEvent
func logosEvent(callerRet C.int, msg *C.char, length C.size_t, userData unsafe.Pointer) {
	h := cgo.Handle(uintptr(userData))
	fn, ok := h.Value().(EventHandler)
	if !ok {
		return
	}
	fn(int(callerRet), goStringN(msg, length))
}

// pendingFrom resolves the cgo.Handle a callback was handed back to its call.
func pendingFrom(userData unsafe.Pointer) (*pending, bool) {
	p, ok := cgo.Handle(uintptr(userData)).Value().(*pending)
	return p, ok
}

// await runs one synchronous entry point and blocks until its callback reports
// a terminal result, returning the reply (on RetOK) or an error built from it.
// invoke receives the cgo.Handle to pass through as userData and returns the
// entry point's immediate return code; a non-zero code means the call was never
// dispatched, so no callback will arrive.
func await(invoke func(ud C.uintptr_t) C.int) (string, error) {
	p := newPending()
	h := cgo.NewHandle(p)
	defer h.Delete()

	if rc := invoke(C.uintptr_t(h)); rc != RetOK {
		return "", fmt.Errorf("liblogosdelivery call was not dispatched (code %d)", int(rc))
	}
	<-p.done
	return p.msg, p.err
}

// New builds a node from a configuration JSON string and returns its handle.
// Creation is asynchronous: this waits for the library to report the node ready
// before returning. The handle must be released with Destroy.
func New(configJSON string) (Handle, error) {
	cCfg := C.CString(configJSON)
	defer C.free(unsafe.Pointer(cCfg))

	p := newPending()
	h := cgo.NewHandle(p)
	defer h.Delete()

	// The constructor's return value is the context handle; the callback
	// reports whether construction actually succeeded. Both are needed: the
	// library hands back the handle immediately but fills the node in on its
	// own thread.
	ctx := C.cGoCreateNode(cCfg, C.uintptr_t(h))
	<-p.done

	if p.err != nil {
		return nil, p.err
	}
	if ctx == nil {
		return nil, errors.New("logosdelivery_create_node returned no context")
	}
	return Handle(ctx), nil
}

// Start starts the node's protocols and Messaging API services.
func Start(h Handle) error {
	_, err := await(func(ud C.uintptr_t) C.int { return C.cGoStartNode(h, ud) })
	return err
}

// Stop stops the node. It can be started again.
func Stop(h Handle) error {
	_, err := await(func(ud C.uintptr_t) C.int { return C.cGoStopNode(h, ud) })
	return err
}

// Destroy releases the node context. Unlike the other entry points it is
// synchronous, and it also drops every event listener registered on the
// context, so h must not be used afterwards.
func Destroy(h Handle) error {
	if rc := C.logosdelivery_destroy(h); rc != RetOK {
		return fmt.Errorf("logosdelivery_destroy failed (code %d)", int(rc))
	}
	return nil
}

// Subscribe subscribes the node to a content topic.
func Subscribe(h Handle, contentTopic string) error {
	cTopic := C.CString(contentTopic)
	defer C.free(unsafe.Pointer(cTopic))
	_, err := await(func(ud C.uintptr_t) C.int { return C.cGoSubscribe(h, cTopic, ud) })
	return err
}

// Unsubscribe unsubscribes the node from a content topic.
func Unsubscribe(h Handle, contentTopic string) error {
	cTopic := C.CString(contentTopic)
	defer C.free(unsafe.Pointer(cTopic))
	_, err := await(func(ud C.uintptr_t) C.int { return C.cGoUnsubscribe(h, cTopic, ud) })
	return err
}

// Send sends a message (JSON: {contentTopic, payload(base64), ephemeral}) and
// returns the request id used to correlate later send events.
func Send(h Handle, messageJSON string) (requestID string, err error) {
	cMsg := C.CString(messageJSON)
	defer C.free(unsafe.Pointer(cMsg))
	return await(func(ud C.uintptr_t) C.int { return C.cGoSend(h, cMsg, ud) })
}

// listeners keeps the cgo.Handle backing each registered listener alive until
// it is removed, keyed by the context and listener id that identify it.
var (
	listenersMu sync.Mutex
	listeners   = make(map[listenerKey]cgo.Handle)
)

type listenerKey struct {
	h  Handle
	id ListenerID
}

// AddEventListener registers fn to receive the named event for the node, and
// returns the id that removes it again. Event names are the library's wire
// names, e.g. "onMessageReceived". Register before Start so no event is missed.
func AddEventListener(h Handle, eventName string, fn EventHandler) (ListenerID, error) {
	cName := C.CString(eventName)
	defer C.free(unsafe.Pointer(cName))

	handle := cgo.NewHandle(fn)
	id := ListenerID(C.cGoAddEventListener(h, cName, C.uintptr_t(handle)))
	if id == 0 {
		handle.Delete()
		return 0, fmt.Errorf("failed to add %q event listener: invalid context", eventName)
	}

	listenersMu.Lock()
	listeners[listenerKey{h, id}] = handle
	listenersMu.Unlock()
	return id, nil
}

// RemoveEventListener removes a listener previously added with
// AddEventListener. Removing an unknown listener is an error.
func RemoveEventListener(h Handle, id ListenerID) error {
	key := listenerKey{h, id}

	listenersMu.Lock()
	handle, known := listeners[key]
	delete(listeners, key)
	listenersMu.Unlock()

	rc := C.logosdelivery_remove_event_listener(h, C.uint64_t(id))
	if known {
		handle.Delete()
	}
	if rc != RetOK {
		return fmt.Errorf("failed to remove event listener %d", uint64(id))
	}
	return nil
}