//go:build js && wasm

package jsws

import (
	"errors"
	"fmt"
	"net"
	"sync"
	"syscall/js"
	"time"
)

const maxStreamQueue = 8192

// StreamConn adapts a browser WebSocket to net.Conn. It sends each Write as
// one binary message and reads message payloads as a byte stream.
type StreamConn struct {
	ws js.Value

	mu     sync.Mutex
	queue  [][]byte
	buf    []byte
	closed bool
	err    error
	notify chan struct{}

	funcs  []js.Func
	local  net.Addr
	remote net.Addr
}

// DialStream opens a binary WebSocket and returns it as a net.Conn.
func DialStream(rawURL string) (*StreamConn, error) {
	ctor := js.Global().Get("WebSocket")
	if ctor.Type() == js.TypeUndefined {
		return nil, errors.New("jsws: WebSocket is not available")
	}
	c := &StreamConn{
		notify: make(chan struct{}, 1),
		local:  streamAddr("stream:local"),
		remote: streamAddr(rawURL),
	}
	ws, err := newWebSocket(ctor, rawURL)
	if err != nil {
		return nil, err
	}
	c.ws = ws
	ws.Set("binaryType", "arraybuffer")

	opened := make(chan struct{})
	failed := make(chan error, 1)
	var openOnce sync.Once
	onOpen := js.FuncOf(func(this js.Value, args []js.Value) any {
		openOnce.Do(func() { close(opened) })
		return nil
	})
	onMessage := js.FuncOf(func(this js.Value, args []js.Value) any {
		if len(args) == 0 {
			return nil
		}
		data := args[0].Get("data")
		if data.Type() != js.TypeObject {
			return nil
		}
		view := js.Global().Get("Uint8Array").New(data)
		buf := make([]byte, view.Length())
		js.CopyBytesToGo(buf, view)
		c.push(buf)
		return nil
	})
	onClose := js.FuncOf(func(this js.Value, args []js.Value) any {
		err := errors.New("jsws: stream closed")
		c.fail(err)
		select {
		case failed <- err:
		default:
		}
		return nil
	})
	onError := js.FuncOf(func(this js.Value, args []js.Value) any {
		err := errors.New("jsws: stream error")
		c.fail(err)
		select {
		case failed <- err:
		default:
		}
		return nil
	})
	c.funcs = []js.Func{onOpen, onMessage, onClose, onError}
	ws.Set("onopen", onOpen)
	ws.Set("onmessage", onMessage)
	ws.Set("onclose", onClose)
	ws.Set("onerror", onError)

	select {
	case <-opened:
		return c, nil
	case err := <-failed:
		return nil, err
	case <-time.After(15 * time.Second):
		func() {
			defer func() { recover() }()
			ws.Call("close")
		}()
		return nil, errors.New("jsws: stream connect timeout")
	}
}

func (c *StreamConn) push(buf []byte) {
	c.mu.Lock()
	if c.closed {
		c.mu.Unlock()
		return
	}
	if len(c.queue) >= maxStreamQueue {
		c.closed = true
		c.err = errors.New("jsws: stream queue overflow")
		c.mu.Unlock()
		c.signal()
		return
	}
	c.queue = append(c.queue, buf)
	c.mu.Unlock()
	c.signal()
}

func (c *StreamConn) fail(err error) {
	c.mu.Lock()
	if !c.closed {
		c.closed = true
		c.err = err
	}
	c.mu.Unlock()
	c.signal()
}

func (c *StreamConn) signal() {
	select {
	case c.notify <- struct{}{}:
	default:
	}
}

// Read implements net.Conn. It waits for WebSocket messages; a close or error
// event signals the same channel, so a dead socket does not leave the caller
// asleep. The P2P read deadline is advisory here because the browser has no
// portable socket timeout; the bridge's idle bound closes the stream.
func (c *StreamConn) Read(p []byte) (int, error) {
	if len(p) == 0 {
		return 0, nil
	}
	for {
		c.mu.Lock()
		if len(c.buf) > 0 {
			n := copy(p, c.buf)
			c.buf = c.buf[n:]
			c.mu.Unlock()
			return n, nil
		}
		if len(c.queue) > 0 {
			c.buf = c.queue[0]
			c.queue = c.queue[1:]
			c.mu.Unlock()
			continue
		}
		if c.closed {
			err := c.err
			c.mu.Unlock()
			if err == nil {
				err = errors.New("jsws: closed")
			}
			return 0, err
		}
		c.mu.Unlock()
		<-c.notify
	}
}

// Write implements net.Conn.
func (c *StreamConn) Write(p []byte) (n int, err error) {
	c.mu.Lock()
	closed := c.closed
	c.mu.Unlock()
	if closed {
		return 0, errors.New("jsws: closed")
	}
	defer func() {
		if r := recover(); r != nil {
			n, err = 0, fmt.Errorf("jsws: send: %v", r)
		}
	}()
	arr := js.Global().Get("Uint8Array").New(len(p))
	js.CopyBytesToJS(arr, p)
	c.ws.Call("send", arr)
	return len(p), nil
}

// Close implements net.Conn.
func (c *StreamConn) Close() error {
	c.mu.Lock()
	already := c.closed
	c.closed = true
	c.mu.Unlock()
	if !already {
		func() {
			defer func() { recover() }()
			c.ws.Call("close")
		}()
	}
	c.signal()
	return nil
}

// LocalAddr implements net.Conn.
func (c *StreamConn) LocalAddr() net.Addr { return c.local }

// RemoteAddr implements net.Conn.
func (c *StreamConn) RemoteAddr() net.Addr { return c.remote }

// SetDeadline implements net.Conn.
func (c *StreamConn) SetDeadline(t time.Time) error { return nil }

// SetReadDeadline implements net.Conn.
func (c *StreamConn) SetReadDeadline(t time.Time) error { return nil }

// SetWriteDeadline implements net.Conn.
func (c *StreamConn) SetWriteDeadline(t time.Time) error { return nil }

type streamAddr string

func (a streamAddr) Network() string { return "ws" }
func (a streamAddr) String() string  { return string(a) }