11package blockchain
22
33import (
4+ "bufio"
45 "bytes"
56 "context"
67 "encoding/json"
78 "fmt"
89 "io"
910 "net/http"
11+ "strings"
12+ "sync"
1013
1114 "github.com/functionland/go-fula/wap/pkg/wifi"
1215 "github.com/libp2p/go-libp2p/core/network"
@@ -20,6 +23,51 @@ const (
2023 GB = 1024 * MB
2124)
2225
26+ type StreamBuffer struct {
27+ Chunks chan string
28+ closed bool
29+ mu sync.Mutex
30+ closeOnce sync.Once
31+ err error
32+ }
33+
34+ func NewStreamBuffer () * StreamBuffer {
35+ return & StreamBuffer {
36+ Chunks : make (chan string , 100 ), // Buffered channel to prevent blocking
37+ }
38+ }
39+
40+ func (b * StreamBuffer ) GetChunk () (string , error ) {
41+ chunk , ok := <- b .Chunks
42+ if ! ok {
43+ return "" , b .err
44+ }
45+ return chunk , nil
46+ }
47+
48+ func (b * StreamBuffer ) AddChunk (chunk string ) {
49+ b .mu .Lock ()
50+ defer b .mu .Unlock ()
51+ if ! b .closed {
52+ b .Chunks <- chunk
53+ }
54+ }
55+
56+ func (b * StreamBuffer ) Close (err error ) {
57+ b .closeOnce .Do (func () {
58+ b .mu .Lock ()
59+ defer b .mu .Unlock ()
60+ b .closed = true
61+ close (b .Chunks )
62+ })
63+ }
64+
65+ func (b * StreamBuffer ) IsClosed () bool {
66+ b .mu .Lock ()
67+ defer b .mu .Unlock ()
68+ return b .closed
69+ }
70+
2371func (bl * FxBlockchain ) BloxFreeSpace (ctx context.Context , to peer.ID ) ([]byte , error ) {
2472 if bl .allowTransientConnection {
2573 ctx = network .WithUseTransient (ctx , "fx.blockchain" )
@@ -245,6 +293,85 @@ func (bl *FxBlockchain) FetchContainerLogs(ctx context.Context, to peer.ID, r wi
245293 }
246294}
247295
296+ func (bl * FxBlockchain ) ChatWithAI (ctx context.Context , to peer.ID , r wifi.ChatWithAIRequest ) (* StreamBuffer , error ) {
297+ if bl .allowTransientConnection {
298+ ctx = network .WithUseTransient (ctx , "fx.blockchain" )
299+ }
300+
301+ var buf bytes.Buffer
302+ if err := json .NewEncoder (& buf ).Encode (r ); err != nil {
303+ return nil , fmt .Errorf ("failed to encode request: %w" , err )
304+ }
305+
306+ req , err := http .NewRequestWithContext (ctx , http .MethodPost , "http://" + to .String ()+ ".invalid/" + actionChatWithAI , & buf )
307+ if err != nil {
308+ return nil , fmt .Errorf ("failed to create request: %w" , err )
309+ }
310+
311+ resp , err := bl .c .Do (req )
312+ if err != nil {
313+ return nil , fmt .Errorf ("failed to send request: %w" , err )
314+ }
315+
316+ if resp .StatusCode != http .StatusOK {
317+ defer resp .Body .Close ()
318+ bodyBytes , _ := io .ReadAll (resp .Body )
319+ return nil , fmt .Errorf ("unexpected response: %d; body: %s" , resp .StatusCode , string (bodyBytes ))
320+ }
321+
322+ buffer := NewStreamBuffer ()
323+
324+ go func () {
325+ defer func () {
326+ resp .Body .Close ()
327+ if ! buffer .IsClosed () {
328+ buffer .Close (nil )
329+ }
330+ }()
331+
332+ reader := bufio .NewReader (resp .Body )
333+ var accum []byte
334+
335+ for {
336+ select {
337+ case <- ctx .Done ():
338+ buffer .Close (fmt .Errorf ("request canceled: %w" , ctx .Err ()))
339+ return
340+ default :
341+ line , err := reader .ReadBytes ('\n' )
342+ if err != nil {
343+ if err == io .EOF {
344+ if len (accum ) > 0 {
345+ buffer .AddChunk (string (accum ))
346+ }
347+ buffer .Close (nil )
348+ } else {
349+ buffer .Close (fmt .Errorf ("error reading response: %w" , err ))
350+ }
351+ return
352+ }
353+
354+ // Check for completion marker
355+ if bytes .HasPrefix (line , []byte ("!COMPLETION!" )) {
356+ buffer .Close (nil )
357+ return
358+ }
359+
360+ accum = append (accum , line ... )
361+
362+ // Try to parse as JSON to verify chunk completeness
363+ var temp interface {}
364+ if json .Unmarshal (accum , & temp ) == nil {
365+ buffer .AddChunk (string (accum ))
366+ accum = accum [:0 ] // Reset accumulator
367+ }
368+ }
369+ }
370+ }()
371+
372+ return buffer , nil
373+ }
374+
248375func (bl * FxBlockchain ) FindBestAndTargetInLogs (ctx context.Context , to peer.ID , r wifi.FindBestAndTargetInLogsRequest ) ([]byte , error ) {
249376
250377 if bl .allowTransientConnection {
@@ -468,6 +595,98 @@ func (bl *FxBlockchain) handleFetchContainerLogs(ctx context.Context, from peer.
468595
469596}
470597
598+ func (bl * FxBlockchain ) handleChatWithAI (ctx context.Context , from peer.ID , w http.ResponseWriter , r * http.Request ) {
599+ log := log .With ("action" , actionChatWithAI , "from" , from )
600+
601+ // Decode the incoming request
602+ var req wifi.ChatWithAIRequest
603+ if err := json .NewDecoder (r .Body ).Decode (& req ); err != nil {
604+ log .Error ("failed to decode request: %v" , err )
605+ http .Error (w , "failed to decode request" , http .StatusBadRequest )
606+ return
607+ }
608+
609+ // Set up headers for streaming response
610+ w .Header ().Set ("Content-Type" , "application/json" )
611+ w .WriteHeader (http .StatusOK )
612+
613+ flusher , ok := w .(http.Flusher )
614+ if ! ok {
615+ http .Error (w , "Streaming not supported" , http .StatusInternalServerError )
616+ return
617+ }
618+
619+ // Fetch AI response using FetchAIResponse
620+ chunks , err := wifi .FetchAIResponse (ctx , req .AIModel , req .UserMessage )
621+ if err != nil {
622+ log .Error ("error in fetchAIResponse: %v" , err )
623+ http .Error (w , fmt .Sprintf ("Error fetching AI response: %v" , err ), http .StatusInternalServerError )
624+ return
625+ }
626+
627+ log .Debugw ("Streaming AI response started" , "ai_model" , req .AIModel , "user_message" , req .UserMessage )
628+ defer log .Debugw ("Streaming AI response ended" , "ai_model" , req .AIModel , "user_message" , req .UserMessage )
629+
630+ var buffer string // Buffer to store incomplete chunks
631+
632+ for {
633+ select {
634+ case <- ctx .Done (): // Handle client disconnect or cancellation
635+ log .Warn ("client disconnected" )
636+ return
637+ case chunk , ok := <- chunks :
638+ if ! ok {
639+ return // Channel closed
640+ }
641+
642+ chunk = strings .TrimSpace (chunk ) // Remove leading/trailing whitespace
643+
644+ if chunk == "" { // Skip empty chunks
645+ continue
646+ }
647+
648+ buffer += chunk // Append chunk to buffer
649+
650+ var parsedChunk struct {
651+ ID string `json:"id"`
652+ Object string `json:"object"`
653+ Choices []struct {
654+ Delta struct {
655+ Content string `json:"content"`
656+ } `json:"delta"`
657+ } `json:"choices"`
658+ }
659+
660+ if err := json .Unmarshal ([]byte (buffer ), & parsedChunk ); err != nil {
661+ log .Error ("failed to parse chunk: %v" , err )
662+ continue // Wait for more data to complete the JSON object
663+ }
664+
665+ buffer = "" // Clear buffer after successful parsing
666+
667+ var newContent string
668+ for _ , choice := range parsedChunk .Choices {
669+ newContent += choice .Delta .Content
670+ }
671+
672+ newContent = strings .TrimSpace (newContent ) // Remove whitespace
673+
674+ response := wifi.ChatWithAIResponse {
675+ Status : true ,
676+ Msg : newContent ,
677+ }
678+ log .Debugw ("Streaming AI response chunk" , "chunk" , newContent )
679+
680+ if err := json .NewEncoder (w ).Encode (response ); err != nil {
681+ log .Error ("failed to write response: %v" , err )
682+ http .Error (w , fmt .Sprintf ("Error writing response: %v" , err ), http .StatusInternalServerError )
683+ return
684+ }
685+ flusher .Flush () // Flush each chunk to ensure real-time streaming
686+ }
687+ }
688+ }
689+
471690func (bl * FxBlockchain ) handleFindBestAndTargetInLogs (ctx context.Context , from peer.ID , w http.ResponseWriter , r * http.Request ) {
472691 log := log .With ("action" , actionFindBestAndTargetInLogs , "from" , from )
473692
0 commit comments