progress (reporting)

Author: danja

.claude/settings.local.json

@@ -30,7 +30,8 @@
       "Bash(unset:*)",
       "Bash(LOG_LEVEL=debug ./trans link-finder)",
       "Bash(cat:*)",
-      "Bash(LOG_LEVEL=trace ./trans:*)"
+      "Bash(LOG_LEVEL=trace ./trans:*)",
+      "Bash(\"http://localhost:3030/test/query?query=PREFIX%20bm%3A%20%3Chttp%3A%2F%2Fpurl.org%2Fstuff%2Fbm%2F%3E%0ASELECT%20%28COUNT%28%3Fbookmark%29%20AS%20%3Fcount%29%20FROM%20%3Chttp%3A%2F%2Fhyperdata.it%2Fcontent%3E%20WHERE%20%7B%20%3Fbookmark%20a%20bm%3ABookmark%20%7D\")"
     ],
     "deny": []
   },

src/apps/link-finder/about.md

@@ -40,17 +40,23 @@ If no entry exists for the bm:target, then this new one is created. Otherwise th
         bm:status "200"^^xsd:integer .
 ```
 
---
-PREFIX : <http://purl.org/stuff/transmissions/>
+## Verification Query
+
+Count bookmarks in the store:
+
+```sparql
 PREFIX bm: <http://purl.org/stuff/bm/>
 
-INSERT DATA {
-      GRAPH <{{graph}}> {
-  <http://purl.org/stuff/instance/eprxly1u> a bm:Bookmark ;
-        bm:target <https://tensegrity.it/> ;
-        bm:agent <http://purl.org/stuff/agent/transmissions> ;
-        bm:created "2025-09-28" ;
-        bm:title "Tensegrity" ;
-        bm:status "404"^^xsd:integer .
-}
+SELECT (COUNT(?bookmark) AS ?count)
+FROM <http://hyperdata.it/content>
+WHERE {
+  ?bookmark a bm:Bookmark .
 }
+```
+
+Or via curl:
+
+```sh
+curl -s -H "Accept: text/plain" --user admin:admin123 \
+  "http://localhost:3030/test/query?query=PREFIX%20bm%3A%20%3Chttp%3A%2F%2Fpurl.org%2Fstuff%2Fbm%2F%3E%0ASELECT%20%28COUNT%28%3Fbookmark%29%20AS%20%3Fcount%29%20FROM%20%3Chttp%3A%2F%2Fhyperdata.it%2Fcontent%3E%20WHERE%20%7B%20%3Fbookmark%20a%20bm%3ABookmark%20%7D"
+```

src/apps/link-finder/config.ttl

@@ -8,8 +8,8 @@
 # ns.trn.sourceFile
 
 :origin a :ConfigSet ;
-    :sourceFile "test.md" .
-   # :targetField "links" .
+    :sourceFile "workflowy_2025-09-21.md" .
+   # :sourceFile "test.md" . 
 
 :listField a :ConfigSet ;
     :remove "true" ;
@@ -20,6 +20,9 @@
     :pp1   :pre     "currentItem.url" ;
             :post    "url"  .
 
+:httpSettings a :ConfigSet ;
+    :timeout "10000" .
+
 :normalizeURL a :ConfigSet ;
     :inputField "resource.target" ;
     :outputField "resource.target" ;
@@ -63,7 +66,7 @@
 
 :updateBookmark a :ConfigSet ;
 # :loglevel "debug" ;
-    :delay "100" ;
+    :delay "500" ;
     :dataBlock "resource" ;
     :templateFilename "data/update-bookmark.njk" ;
     :noCache "true" ; # always reload endpoints

src/processors/flow/ForEach.js

@@ -92,6 +92,7 @@ class ForEach extends SlowableProcessor {
         const reduced = current
 
         logger.debug(`    reduced.length = ${reduced.length}`)
+        logger.log(`ForEach: Processing ${reduced.length} items`)
 
         //    logger.v(reduced)
 
@@ -110,6 +111,9 @@ class ForEach extends SlowableProcessor {
             message = JSONUtils.remove(message, forEach)
         }
         message.done = false
+        const totalItems = reduced.length
+        const progressInterval = Math.max(1, Math.floor(totalItems / 20)) // Report every 5%
+
         for (const item of reduced) {
 
             if (limitString && ++counter > limit) continue
@@ -124,11 +128,19 @@ class ForEach extends SlowableProcessor {
             clonedMessage.foreach = undefined
 
             logger.log(`ForEach: Emitting message for item: ${JSON.stringify(item)}`)
-            clonedMessage.eachCounter = this.eachCounter++
+            clonedMessage.eachCounter = this.eachCounter
+
+            // Progress logging
+            if (this.eachCounter % progressInterval === 0 || this.eachCounter === totalItems - 1) {
+                const percent = Math.round((this.eachCounter / totalItems) * 100)
+                logger.log(`ForEach: Progress ${this.eachCounter + 1}/${totalItems} (${percent}%)`)
+            }
+
+            this.eachCounter++
             this.emit('message', clonedMessage)
         }
         message.done = true
-        logger.log('ForEach: all messages dispatched')
+        logger.log(`ForEach: All ${totalItems} messages dispatched`)
 
         this.emit('message', message)
     }

src/processors/http/HttpClient.js

@@ -89,8 +89,18 @@ class HttpClient extends Processor {
         const method = super.getProperty(ns.trn.method, 'GET').toUpperCase()
         const headers = super.getProperty(ns.trn.headers, {})
         const body = super.getProperty(ns.trn.body, null)
+        const timeout = super.getProperty(ns.trn.timeout, 30000) // Default 30s timeout
 
         const options = { method, headers }
+
+        // Add timeout using AbortController
+        if (timeout) {
+            const controller = new AbortController()
+            options.signal = controller.signal
+            options.controller = controller
+            options.timeout = parseInt(timeout)
+        }
+
         if (body && (method === 'POST' || method === 'PUT' || method === 'PATCH')) {
             options.body = typeof body === 'string' ? body : JSON.stringify(body)
             if (!headers['Content-Type']) {
@@ -107,7 +117,24 @@ class HttpClient extends Processor {
      * @returns {Promise<Response>}
      */
     async _sendRequest(requestOptions) {
-        const { url, ...options } = requestOptions
+        const { url, timeout, controller, ...options } = requestOptions
+
+        // Handle timeout
+        if (timeout && controller) {
+            const timeoutId = setTimeout(() => {
+                controller.abort()
+            }, timeout)
+
+            try {
+                const response = await fetch(url, options)
+                clearTimeout(timeoutId)
+                return response
+            } catch (error) {
+                clearTimeout(timeoutId)
+                throw error
+            }
+        }
+
         return fetch(url, options)
     }