Skip to content

Commit a9902be

Browse files
maximbilanmaximbilan
authored
Refactoring (#247)
1 parent c977ed4 commit a9902be

1 file changed

Lines changed: 141 additions & 99 deletions

File tree

‎internal/app/therapy.go‎

Lines changed: 141 additions & 99 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,21 @@ import (
88
"log"
99
"net/http"
1010
"os"
11-
"strings"
11+
"strings"
1212
"time"
1313

1414
"github.com/google/uuid"
1515
)
1616

17+
// Internal configuration for therapy session calls
18+
type therapyConfig struct {
19+
baseURL string
20+
token string
21+
userID string
22+
sessionID string
23+
locale string
24+
}
25+
1726
// Start therapy session
1827
func startTherapySession(session *Session) {
1928
setOutputText("start_therapy_session", session)
@@ -25,75 +34,113 @@ func startTherapySession(session *Session) {
2534
// HTTP client to therapy session endpoint
2635
func callTherapySessionEndpoint(text string, session *Session) *string {
2736
//coverage:ignore
28-
// Resolve base URL and token
37+
cfg := buildTherapyConfig(session)
38+
if cfg == nil {
39+
return nil
40+
}
41+
42+
client := buildTherapyHTTPClient()
43+
44+
if !initTherapySession(client, cfg) {
45+
return nil
46+
}
47+
48+
return sendTherapyMessage(client, cfg, text)
49+
}
50+
51+
// Relay a user message to the therapy session backend and append the reply
52+
func relayTherapyMessage(text string, session *Session) {
53+
//coverage:ignore
54+
// Send immediate typing acknowledgement is already enabled via IsTyping
55+
reply := callTherapySessionEndpoint(text, session)
56+
if reply != nil && *reply != "" {
57+
setOutputRawText(*reply, session)
58+
}
59+
}
60+
61+
// Build an HTTP client configured for therapy session calls
62+
func buildTherapyHTTPClient() *http.Client {
63+
return &http.Client{Timeout: 120 * time.Second}
64+
}
65+
66+
// Build configuration from environment and session; ensures a session ID exists
67+
func buildTherapyConfig(session *Session) *therapyConfig {
2968
baseURL := os.Getenv("CAPY_THERAPY_SESSION_URL")
3069
if baseURL == "" {
3170
log.Printf("[TherapySession] missing CAPY_THERAPY_SESSION_URL")
3271
return nil
3372
}
73+
3474
token := os.Getenv("CAPY_AGENT_TOKEN")
3575
if token == "" {
3676
log.Printf("[TherapySession] missing CAPY_AGENT_TOKEN")
3777
return nil
3878
}
3979

40-
// Ensure therapy session identifier on the user entity
4180
if session.User.TherapySessionId == nil || *session.User.TherapySessionId == "" {
4281
newID := uuid.NewString()
4382
session.User.TherapySessionId = &newID
4483
}
4584

4685
userID := session.User.ID
47-
therapySessionID := *session.User.TherapySessionId
86+
sessionID := *session.User.TherapySessionId
4887
locale := session.Locale().String()
4988

50-
client := &http.Client{Timeout: 120 * time.Second}
89+
return &therapyConfig{
90+
baseURL: baseURL,
91+
token: token,
92+
userID: userID,
93+
sessionID: sessionID,
94+
locale: locale,
95+
}
96+
}
5197

52-
// 1) Create/init the therapy session
53-
initURL := fmt.Sprintf("%s/apps/capymind_agent/users/%s/sessions/%s", baseURL, userID, therapySessionID)
98+
// Initialize or validate the therapy session on the backend
99+
func initTherapySession(client *http.Client, cfg *therapyConfig) bool {
100+
initURL := fmt.Sprintf("%s/apps/capymind_agent/users/%s/sessions/%s", cfg.baseURL, cfg.userID, cfg.sessionID)
54101
initBody := map[string]any{
55102
"state": map[string]any{
56-
"preferred_language": locale,
103+
"preferred_language": cfg.locale,
57104
},
58105
}
59106
initBodyBytes, _ := json.Marshal(initBody)
60-
initReq, err := http.NewRequest("POST", initURL, bytes.NewBuffer(initBodyBytes))
107+
108+
req, err := http.NewRequest("POST", initURL, bytes.NewBuffer(initBodyBytes))
61109
if err != nil {
62110
log.Printf("[TherapySession] init request build error: %v", err)
63-
return nil
111+
return false
64112
}
65-
initReq.Header.Set("Authorization", "Bearer "+token)
66-
initReq.Header.Set("Content-Type", "application/json")
67-
68-
initResp, err := client.Do(initReq)
69-
if err != nil {
70-
log.Printf("[TherapySession] init request error: %v", err)
71-
return nil
72-
}
73-
defer initResp.Body.Close()
74-
proceed := false
75-
if initResp.StatusCode >= 200 && initResp.StatusCode < 300 {
76-
proceed = true
77-
} else {
78-
body, _ := io.ReadAll(initResp.Body)
79-
// Allow existing session scenario to proceed
80-
if initResp.StatusCode == 400 && strings.Contains(string(body), "Session already exists") {
81-
log.Printf("[TherapySession] init session exists, proceeding: %s", therapySessionID)
82-
proceed = true
83-
} else {
84-
log.Printf("[TherapySession] init non-2xx: %d body=%s", initResp.StatusCode, string(body))
85-
}
86-
}
87-
if !proceed {
88-
return nil
89-
}
90-
91-
// 2) Send user message via run_sse
92-
runURL := fmt.Sprintf("%s/run_sse", baseURL)
113+
req.Header.Set("Authorization", "Bearer "+cfg.token)
114+
req.Header.Set("Content-Type", "application/json")
115+
116+
resp, err := client.Do(req)
117+
if err != nil {
118+
log.Printf("[TherapySession] init request error: %v", err)
119+
return false
120+
}
121+
defer resp.Body.Close()
122+
123+
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
124+
return true
125+
}
126+
127+
body, _ := io.ReadAll(resp.Body)
128+
if resp.StatusCode == 400 && strings.Contains(string(body), "Session already exists") {
129+
log.Printf("[TherapySession] init session exists, proceeding: %s", cfg.sessionID)
130+
return true
131+
}
132+
133+
log.Printf("[TherapySession] init non-2xx: %d body=%s", resp.StatusCode, string(body))
134+
return false
135+
}
136+
137+
// Send the user's message to the run_sse endpoint and parse the reply
138+
func sendTherapyMessage(client *http.Client, cfg *therapyConfig, text string) *string {
139+
runURL := fmt.Sprintf("%s/run_sse", cfg.baseURL)
93140
runBody := map[string]any{
94141
"app_name": "capymind_agent",
95-
"user_id": userID,
96-
"session_id": therapySessionID,
142+
"user_id": cfg.userID,
143+
"session_id": cfg.sessionID,
97144
"new_message": map[string]any{
98145
"role": "user",
99146
"parts": []map[string]string{
@@ -103,83 +150,78 @@ func callTherapySessionEndpoint(text string, session *Session) *string {
103150
"streaming": false,
104151
}
105152
runBodyBytes, _ := json.Marshal(runBody)
106-
runReq, err := http.NewRequest("POST", runURL, bytes.NewBuffer(runBodyBytes))
153+
154+
req, err := http.NewRequest("POST", runURL, bytes.NewBuffer(runBodyBytes))
107155
if err != nil {
108156
log.Printf("[TherapySession] run request build error: %v", err)
109157
return nil
110158
}
111-
runReq.Header.Set("Authorization", "Bearer "+token)
112-
runReq.Header.Set("Content-Type", "application/json")
159+
req.Header.Set("Authorization", "Bearer "+cfg.token)
160+
req.Header.Set("Content-Type", "application/json")
113161

114-
runResp, err := client.Do(runReq)
162+
resp, err := client.Do(req)
115163
if err != nil {
116164
log.Printf("[TherapySession] run request error: %v", err)
117165
return nil
118166
}
119-
defer runResp.Body.Close()
120-
runRespBody, err := io.ReadAll(runResp.Body)
167+
defer resp.Body.Close()
168+
169+
respBody, err := io.ReadAll(resp.Body)
121170
if err != nil {
122171
log.Printf("[TherapySession] run read error: %v", err)
123172
return nil
124173
}
125-
if runResp.StatusCode < 200 || runResp.StatusCode >= 300 {
126-
log.Printf("[TherapySession] run non-2xx: %d body=%s", runResp.StatusCode, string(runRespBody))
174+
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
175+
log.Printf("[TherapySession] run non-2xx: %d body=%s", resp.StatusCode, string(respBody))
176+
return nil
177+
}
178+
179+
respStr := string(respBody)
180+
if respStr == "" {
127181
return nil
128182
}
129-
respStr := string(runRespBody)
130-
if respStr == "" {
131-
return nil
132-
}
133-
134-
// Try to extract plain text from JSON response
135-
// Support responses that are either raw JSON or lines prefixed with "data: "
136-
extractJSON := func(s string) string {
137-
s = strings.TrimSpace(s)
138-
if strings.HasPrefix(s, "data:") {
139-
// If multiple lines, pick the last data line
140-
lines := strings.Split(s, "\n")
141-
for i := len(lines) - 1; i >= 0; i-- {
142-
line := strings.TrimSpace(lines[i])
143-
if strings.HasPrefix(line, "data:") {
144-
return strings.TrimSpace(strings.TrimPrefix(line, "data:"))
145-
}
146-
}
147-
return strings.TrimSpace(strings.TrimPrefix(lines[len(lines)-1], "data:"))
148-
}
149-
return s
150-
}
151-
152-
type runSseContentPart struct {
153-
Text string `json:"text"`
154-
}
155-
type runSseContent struct {
156-
Parts []runSseContentPart `json:"parts"`
157-
}
158-
type runSseResponse struct {
159-
Content runSseContent `json:"content"`
160-
}
161-
162-
jsonCandidate := extractJSON(respStr)
163-
var parsed runSseResponse
164-
if err := json.Unmarshal([]byte(jsonCandidate), &parsed); err == nil {
165-
if len(parsed.Content.Parts) > 0 && parsed.Content.Parts[0].Text != "" {
166-
onlyText := parsed.Content.Parts[0].Text
167-
return &onlyText
168-
}
169-
}
170-
171-
// Fallback: return body as-is
172-
return &respStr
183+
return parseRunResponse(respStr)
173184
}
174185

175-
// Relay a user message to the therapy session backend and append the reply
176-
func relayTherapyMessage(text string, session *Session) {
177-
//coverage:ignore
178-
// Send immediate typing acknowledgement is already enabled via IsTyping
179-
reply := callTherapySessionEndpoint(text, session)
180-
if reply != nil && *reply != "" {
181-
setOutputRawText(*reply, session)
186+
// Parse a run_sse HTTP response body, extracting plain text if present
187+
func parseRunResponse(respStr string) *string {
188+
extractJSON := func(s string) string {
189+
s = strings.TrimSpace(s)
190+
if strings.HasPrefix(s, "data:") {
191+
// If multiple lines, pick the last data line
192+
lines := strings.Split(s, "\n")
193+
for i := len(lines) - 1; i >= 0; i-- {
194+
line := strings.TrimSpace(lines[i])
195+
if strings.HasPrefix(line, "data:") {
196+
return strings.TrimSpace(strings.TrimPrefix(line, "data:"))
197+
}
198+
}
199+
return strings.TrimSpace(strings.TrimPrefix(lines[len(lines)-1], "data:"))
200+
}
201+
return s
182202
}
203+
204+
type runSseContentPart struct {
205+
Text string `json:"text"`
206+
}
207+
type runSseContent struct {
208+
Parts []runSseContentPart `json:"parts"`
209+
}
210+
type runSseResponse struct {
211+
Content runSseContent `json:"content"`
212+
}
213+
214+
jsonCandidate := extractJSON(respStr)
215+
var parsed runSseResponse
216+
if err := json.Unmarshal([]byte(jsonCandidate), &parsed); err == nil {
217+
if len(parsed.Content.Parts) > 0 && parsed.Content.Parts[0].Text != "" {
218+
onlyText := parsed.Content.Parts[0].Text
219+
return &onlyText
220+
}
221+
}
222+
223+
// Fallback: return body as-is
224+
return &respStr
183225
}
184226

185227
// End the therapy session and notify the user

0 commit comments

Comments
 (0)