feat(upload): add support for announce_uploaded flag in upload tokens (#55)

Author: msmakouz

src/uploads/_index.yaml

@@ -373,6 +373,7 @@ entries:
     imports:
       upload_repo: userspace.uploads:upload_repo
       upload_type: userspace.uploads:upload_type
+      pipeline_lib: userspace.uploads:pipeline_lib
       uploads_resources: userspace.uploads:uploads_resources
       content_repo: userspace.uploads:content_repo
 

src/uploads/pipeline_lib.lua

@@ -78,8 +78,7 @@ function pipeline_lib.merge_metadata(existing_metadata, new_metadata)
     return merged
 end
 
--- Invoke upload completion token callback (fire-and-forget)
-function pipeline_lib.invoke_upload_token(upload, success, error_msg)
+function pipeline_lib.invoke_upload_token(upload, status, error_msg)
     if not upload or not upload.metadata or not upload.metadata.__upload_token then
         return
     end
@@ -90,7 +89,11 @@ function pipeline_lib.invoke_upload_token(upload, success, error_msg)
         return
     end
 
-    local func_id = tostring(success and payload.function_id or payload.on_error_id)
+    if status == STATUS.UPLOADED and not payload.announce_uploaded then
+        return
+    end
+
+    local func_id = tostring(status == STATUS.ERROR and payload.on_error_id or payload.function_id)
     if not func_id or func_id == "nil" then
         return
     end
@@ -118,7 +121,7 @@ function pipeline_lib.invoke_upload_token(upload, success, error_msg)
 
     local _, call_err = executor:call(tostring(func_id), {
         upload_id = upload.uuid,
-        status = success and "completed" or "error",
+        status = status,
         error = error_msg,
         mime_type = upload.mime_type,
         size = upload.size,
@@ -165,7 +168,7 @@ function pipeline_lib.process_upload(upload)
         -- Notify about error status
         pipeline_lib.notify_status_change(upload, STATUS.ERROR, nil, error_msg)
 
-        pipeline_lib.invoke_upload_token(upload, false, error_msg)
+        pipeline_lib.invoke_upload_token(upload, STATUS.ERROR, error_msg)
 
         return false, error_msg
     end
@@ -190,7 +193,7 @@ function pipeline_lib.process_upload(upload)
         -- Notify about error status
         pipeline_lib.notify_status_change(upload, STATUS.ERROR, nil, error_msg)
 
-        pipeline_lib.invoke_upload_token(upload, false, error_msg)
+        pipeline_lib.invoke_upload_token(upload, STATUS.ERROR, error_msg)
 
         return false, error_msg
     end
@@ -245,7 +248,7 @@ function pipeline_lib.process_upload(upload)
             -- Notify about error status with stage info
             pipeline_lib.notify_status_change(upload, STATUS.ERROR, stage_title, error_msg)
 
-            pipeline_lib.invoke_upload_token(upload, false, error_msg)
+            pipeline_lib.invoke_upload_token(upload, STATUS.ERROR, error_msg)
 
             return false, error_msg
         end
@@ -279,12 +282,12 @@ function pipeline_lib.process_upload(upload)
     -- Notify about completion status
     pipeline_lib.notify_status_change(upload, STATUS.COMPLETED)
 
-    pipeline_lib.invoke_upload_token(upload, true, nil)
+    pipeline_lib.invoke_upload_token(upload, STATUS.COMPLETED)
 
     return true
 end
 
 -- Export constants
 pipeline_lib.STATUS = STATUS
 
-return pipeline_lib
+return pipeline_lib

src/uploads/test/upload_tokens_test.lua

@@ -48,6 +48,22 @@ local function define_tests()
             test.eq(result.actor_scope, "default")
             test.is_nil(result.on_error_id)
             test.is_nil(result.params)
+            -- Default contract: no early announce unless explicitly requested.
+            test.is_nil(result.announce_uploaded)
+        end)
+
+        it("should carry the opt-in announce_uploaded flag", function()
+            local token, err = upload_tokens.pack({
+                function_id = "my.module:handler",
+                actor_id = "user-123",
+                actor_scope = "default",
+                announce_uploaded = true,
+            })
+            test.is_nil(err)
+
+            local result, uerr = upload_tokens.unpack(token :: string)
+            test.is_nil(uerr)
+            test.eq(result.announce_uploaded, true)
         end)
 
         it("should error on missing function_id", function()

src/uploads/upload_lib.lua

@@ -8,6 +8,7 @@ local logger = require("logger")
 
 local upload_repo = require("upload_repo")
 local upload_type = require("upload_type")
+local pipeline_lib = require("pipeline_lib")
 local resources = require("uploads_resources")
 local content_repo = require("content_repo")
 
@@ -287,6 +288,8 @@ function upload_lib.upload_file(user_id: string, file_data: string | stream.Stre
 
     publish_to_queue(upload_uuid)
 
+    pipeline_lib.invoke_upload_token(upload, pipeline_lib.STATUS.UPLOADED)
+
     return upload
 end
 
@@ -433,6 +436,8 @@ function upload_lib.complete_presigned_url(user_id, upload_id, etag, metadata_up
 
     publish_to_queue(upload_id)
 
+    pipeline_lib.invoke_upload_token(upload, pipeline_lib.STATUS.UPLOADED)
+
     return upload
 end
 
@@ -668,6 +673,8 @@ function upload_lib.complete_multipart_upload(user_id, upload_id, parts: {{part_
     upload.updated_at = updated.updated_at
     upload.metadata = metadata
 
+    pipeline_lib.invoke_upload_token(upload, pipeline_lib.STATUS.UPLOADED)
+
     return upload
 end
 

src/uploads/upload_repo.lua

@@ -739,7 +739,9 @@ function upload_repo.complete_s3_upload(user_id, upload_id, etag, key, metadata_
         return nil, "Failed to update upload record: " .. tostring(err)
     end
 
-    return updated
+    upload.status = updated.status
+    upload.updated_at = updated.updated_at
+    return upload
 end
 
 function upload_repo.list_with_filters(options)

src/uploads/upload_tokens.lua

@@ -56,6 +56,7 @@ local function pack(params)
         params = params.params,
         actor_id = params.actor_id,
         actor_scope = params.actor_scope,
+        announce_uploaded = params.announce_uploaded and true or nil,
         issued_at = os.time(),
     }