package app import ( "bufio" "bytes" "context" "encoding/json" "errors" "fmt" "io" "log/slog" "os/exec" "strings" "sync" "time" "git.stormux.org/storm/cthulhu/internal/braille" ) func (worker *brailleWorker) run(ctx context.Context, done chan struct{}) { defer close(done) for ctx.Err() == nil { if err := worker.session(ctx); err != nil && ctx.Err() == nil { worker.logger.Warn("braille helper disconnected", "error", err) } select { case <-ctx.Done(): return case <-time.After(time.Second): } } } func (worker *brailleWorker) session(ctx context.Context) (result error) { defer func() { worker.mu.Lock(); worker.connected = false; worker.mu.Unlock() }() childCtx, cancel := context.WithCancel(ctx) defer cancel() command := exec.CommandContext(childCtx, worker.path, "--client") stdin, err := command.StdinPipe() if err != nil { return err } stdout, err := command.StdoutPipe() if err != nil { return err } var stderr bytes.Buffer command.Stderr = &stderr if err := command.Start(); err != nil { return err } waitDone := make(chan struct{}) var waitErr error go func() { defer close(waitDone); waitErr = command.Wait() }() var ioWorkers sync.WaitGroup // Kill the disposable native owner, close pipes to unblock I/O, then join // every goroutine before permitting reconnect or a new configured session. defer func() { cancel() _ = stdin.Close() _ = stdout.Close() <-waitDone ioWorkers.Wait() if result != nil { if detail := strings.TrimSpace(stderr.String()); detail != "" { result = fmt.Errorf("%w: %s", result, detail) } else if waitErr != nil { result = fmt.Errorf("%w: %v", result, waitErr) } } }() ready := make(chan error, 1) readErrors := make(chan error, 1) // Routing keys retain FIFO order. Backpressure stops this pipe reader at // 16 pending keys; cancellation releases it and retires the session queue. routes := make(chan braille.RoutingEvent, 16) ioWorkers.Add(1) go func() { defer ioWorkers.Done() scanner := bufio.NewScanner(stdout) scanner.Buffer(make([]byte, 1024), 1024) if !scanner.Scan() || scanner.Text() != "ready" { ready <- errors.New("braille helper sent an invalid ready message") return } ready <- nil for scanner.Scan() { var event braille.RoutingEvent if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { readErrors <- errors.New("invalid braille routing event") return } select { case routes <- event: case <-childCtx.Done(): return } } err := scanner.Err() if err == nil { err = io.EOF } readErrors <- err }() timer := time.NewTimer(2 * time.Second) defer timer.Stop() select { case err := <-ready: if err != nil { return err } case <-timer.C: return errors.New("braille helper connection timed out") case <-ctx.Done(): return nil } worker.logger.Info("braille helper connected") worker.mu.Lock() worker.connected = true worker.mu.Unlock() writeError := make(chan error, 1) ioWorkers.Add(1) go func() { defer ioWorkers.Done() encoder := json.NewEncoder(stdin) for { frame := worker.current() if err := encoder.Encode(frame); err != nil { writeError <- err return } // This confirms the helper write, not the physical display. Text is // the uncontracted line; cursor offsets count Unicode characters. if worker.logger.Enabled(context.Background(), slog.LevelDebug) { worker.logger.Debug("sent braille frame", "frame_id", frame.ID, "braille_text", frame.Text, "cursor_offset", frame.Cursor, "contracted", frame.Contract, "table", frame.Table, "word_wrap", frame.WordWrap, "expand_at_cursor", frame.ExpandAtCursor) } select { case <-childCtx.Done(): return case <-worker.wake: } } }() for { select { case <-waitDone: return waitErr case err := <-writeError: return err case err := <-readErrors: return err case event := <-routes: worker.route(childCtx, event) case <-ctx.Done(): return nil } } }