S3 artifact service

.gitignore

@@ -34,6 +34,7 @@ Thumbs.db
 docs/
 plans/
 
+contrib/samples/a2a_basic/bin/
 # Sample build artifacts / local scratch
 contrib/samples/**/bin/
 mkpro_logs.db

core/pom.xml

@@ -66,6 +66,11 @@
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
     </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>s3</artifactId>
+      <version>2.42.31</version>
+    </dependency>
     <dependency>
       <groupId>com.google.genai</groupId>
       <artifactId>google-genai</artifactId>

core/src/main/java/com/google/adk/artifacts/PostgresArtifactService.GUIDE.md

@@ -0,0 +1,75 @@
+# PostgresArtifactService Guide
+
+## What It Does
+
+`PostgresArtifactService` stores artifacts in PostgreSQL and can optionally mirror them to S3
+and emit Kafka events. It is the runtime artifact service used by the application.
+
+Core behavior:
+- Always writes artifact bytes and metadata to Postgres.
+- Uploads bytes to S3 when `S3_BUCKET` is set.
+- Publishes a Kafka event when `use_kafka=true` and `kafka_topic` is set.
+
+## Save Flow (Write Path)
+
+1. Extract bytes and MIME type from the `Part`.
+2. Save to Postgres (data + metadata).
+3. If S3 is enabled, upload bytes to `s3://<bucket>/<key>`.
+4. If Kafka is enabled, publish a JSON event describing the artifact.
+
+## Configuration
+
+### Postgres (required)
+
+Environment variables:
+- `DBURL`
+- `DBUSER`
+- `DBPASSWORD`
+
+### S3 (optional)
+
+Environment variables:
+- `S3_BUCKET` (enables S3 upload)
+- `S3_REGION` (recommended)
+- `S3_ENDPOINT` (optional, for S3-compatible storage)
+- `S3_PATH_STYLE` (`true` for path-style access)
+
+### Kafka (optional)
+
+Properties (via `PropertiesHelper`):
+- `use_kafka=true`
+- `kafka_topic=<topic-name>`
+
+## S3 Object Key Format
+
+```
+<appName>/<userId>/<sessionId>/<filename>/<version>
+```
+
+If `filename` starts with `user:`, the key becomes:
+
+```
+<appName>/<userId>/user/<filename>/<version>
+```
+
+## Kafka Event Payload
+
+Published fields:
+- `type` = `artifact`
+- `appName`, `userId`, `sessionId`, `filename`, `version`
+- `mimeType`
+- `fileUri` (S3 URI)
+- `sizeBytes`
+- `metadata` (only included when non-null)
+
+## Common Checks
+
+- If S3 upload fails, the Postgres write still succeeds (fail-open).
+- If Kafka is disabled or topic is missing, no event is published.
+- Missing `S3_REGION` or invalid AWS credentials will cause S3 failures.
+
+## Related Files
+
+- `PostgresArtifactService.java`
+- `PostgresArtifactStore.java`
+- `S3ArtifactService.java`

core/src/main/java/com/google/adk/artifacts/PostgresArtifactService.java

@@ -16,17 +16,27 @@
 
 package com.google.adk.artifacts;
 
+import com.google.adk.kafka.consumer.KafkaWriter;
 import com.google.adk.store.PostgresArtifactStore;
 import com.google.adk.store.PostgresArtifactStore.ArtifactData;
+import com.google.adk.utils.PropertiesHelper;
 import com.google.common.collect.ImmutableList;
 import com.google.genai.types.Part;
 import io.reactivex.rxjava3.core.Completable;
 import io.reactivex.rxjava3.core.Maybe;
 import io.reactivex.rxjava3.core.Single;
 import io.reactivex.rxjava3.schedulers.Schedulers;
+import java.net.URI;
 import java.sql.SQLException;
 import java.util.List;
+import org.json.JSONObject;
 import org.jspecify.annotations.Nullable;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Exception;
 
 /**
  * A PostgreSQL-backed implementation of the {@link BaseArtifactService}.
@@ -63,7 +73,14 @@
 public final class PostgresArtifactService implements BaseArtifactService {
 
   private static final String DEFAULT_TABLE_NAME = "artifacts";
+  private static final String DEFAULT_MIME_TYPE = "application/octet-stream";
   private final PostgresArtifactStore dbHelper;
+  private final @Nullable S3Client s3Client;
+  private final @Nullable String s3Bucket;
+  private final boolean kafkaEnabled;
+  private final @Nullable String kafkaTopic;
+
+  private final @Nullable String s3BasePath;
 
   /**
    * Creates a new PostgresArtifactService using environment variables for database connection. Uses
@@ -80,6 +97,11 @@ public final class PostgresArtifactService implements BaseArtifactService {
    */
   public PostgresArtifactService() {
     this.dbHelper = PostgresArtifactStore.getInstance(DEFAULT_TABLE_NAME);
+    this.s3Bucket = getenvOrNull("S3_BUCKET");
+    this.s3BasePath = getenvOrNull("S3_BASE_PATH");
+    this.s3Client = s3Bucket != null ? buildS3Client() : null;
+    this.kafkaEnabled = Boolean.parseBoolean(PropertiesHelper.getInstance().getValue("use_kafka"));
+    this.kafkaTopic = PropertiesHelper.getInstance().getValue("kafka_topic");
   }
 
   /**
@@ -95,6 +117,11 @@ public PostgresArtifactService() {
   public PostgresArtifactService(String dbUrl, String dbUser, String dbPassword) {
     this.dbHelper =
         PostgresArtifactStore.createInstance(dbUrl, dbUser, dbPassword, DEFAULT_TABLE_NAME);
+    this.s3Bucket = getenvOrNull("S3_BUCKET");
+    this.s3BasePath = getenvOrNull("S3_BASE_PATH");
+    this.s3Client = s3Bucket != null ? buildS3Client() : null;
+    this.kafkaEnabled = Boolean.parseBoolean(PropertiesHelper.getInstance().getValue("use_kafka"));
+    this.kafkaTopic = PropertiesHelper.getInstance().getValue("kafka_topic");
   }
 
   @Override
@@ -107,10 +134,12 @@ public Single<Integer> saveArtifact(
                 byte[] data = extractBytesFromPart(artifact);
                 String mimeType = extractMimeTypeFromPart(artifact);
 
-                // Save to database without metadata (metadata = null)
-                // Applications should use saveArtifact(..., metadata) if they need custom metadata
-                return dbHelper.saveArtifact(
-                    appName, userId, sessionId, filename, data, mimeType, null);
+                int version =
+                    dbHelper.saveArtifact(
+                        appName, userId, sessionId, filename, data, mimeType, null);
+                persistToS3AndKafka(
+                    appName, userId, sessionId, filename, version, data, mimeType, null);
+                return version;
               } catch (SQLException e) {
                 throw new RuntimeException("Failed to save artifact: " + e.getMessage(), e);
               }
@@ -154,9 +183,12 @@ public Single<Integer> saveArtifact(
                 byte[] data = extractBytesFromPart(artifact);
                 String mimeType = extractMimeTypeFromPart(artifact);
 
-                // Save to database with caller-provided metadata
-                return dbHelper.saveArtifact(
-                    appName, userId, sessionId, filename, data, mimeType, metadata);
+                int version =
+                    dbHelper.saveArtifact(
+                        appName, userId, sessionId, filename, data, mimeType, metadata);
+                persistToS3AndKafka(
+                    appName, userId, sessionId, filename, version, data, mimeType, metadata);
+                return version;
               } catch (SQLException e) {
                 throw new RuntimeException("Failed to save artifact: " + e.getMessage(), e);
               }
@@ -257,13 +289,127 @@ private byte[] extractBytesFromPart(Part part) {
    */
   private String extractMimeTypeFromPart(Part part) {
     if (part.inlineData() != null && part.inlineData().isPresent()) {
-      return part.inlineData().get().mimeType().orElse("application/octet-stream");
+      return part.inlineData().get().mimeType().orElse(DEFAULT_MIME_TYPE);
     }
-    return "application/octet-stream"; // Default fallback
+    return DEFAULT_MIME_TYPE; // Default fallback
+  }
+
+  private void persistToS3AndKafka(
+      String appName,
+      String userId,
+      String sessionId,
+      String filename,
+      int version,
+      byte[] data,
+      String mimeType,
+      @Nullable String metadata) {
+    if (s3Client == null || s3Bucket == null) {
+      return;
+    }
+    String key = buildObjectKey(appName, userId, sessionId, filename, version);
+
+    // Construct an HTTPS URL instead of an s3:// path
+    String region = getenvOrNull("S3_REGION");
+    String fileUri;
+    if (region != null && !region.isBlank()) {
+      fileUri = String.format("https://%s.s3.%s.amazonaws.com/%s", s3Bucket, region, key);
+    } else {
+      // Fallback to global endpoint if region is not set
+      fileUri = String.format("https://%s.s3.amazonaws.com/%s", s3Bucket, key);
+    }
+
+    try {
+      PutObjectRequest.Builder requestBuilder =
+          PutObjectRequest.builder().bucket(s3Bucket).key(key).contentType(mimeType);
+      s3Client.putObject(requestBuilder.build(), RequestBody.fromBytes(data));
+      dbHelper.updateFileUri(appName, userId, sessionId, filename, version, fileUri);
+      publishKafkaEvent(
+          appName, userId, sessionId, filename, version, mimeType, metadata, fileUri, data.length);
+    } catch (S3Exception | SQLException e) {
+      // Fail-open: Postgres save already succeeded; log and continue.
+      System.err.println("⚠️  Failed to persist artifact to S3/Kafka: " + e.getMessage());
+    }
+  }
+
+  private void publishKafkaEvent(
+      String appName,
+      String userId,
+      String sessionId,
+      String filename,
+      int version,
+      String mimeType,
+      @Nullable String metadata,
+      String fileUri,
+      int sizeBytes) {
+    if (!kafkaEnabled || kafkaTopic == null || kafkaTopic.isBlank()) {
+      return;
+    }
+    JSONObject payload = new JSONObject();
+    payload.put("type", "artifact");
+    payload.put("appName", appName);
+    payload.put("userId", userId);
+    payload.put("sessionId", sessionId);
+    payload.put("filename", filename);
+    payload.put("version", version);
+    payload.put("mimeType", mimeType);
+    payload.put("fileUri", fileUri);
+    payload.put("sizeBytes", sizeBytes);
+    if (metadata != null) {
+      payload.put("metadata", metadata);
+    }
+    String key = appName + ":" + userId + ":" + sessionId + ":" + filename + ":" + version;
+    try {
+      KafkaWriter.PublishWithStatus(kafkaTopic, key, payload.toString());
+    } catch (Exception e) {
+      System.err.println("⚠️  Failed to publish artifact event to Kafka: " + e.getMessage());
+    }
+  }
+
+  private String buildObjectKey(
+      String appName, String userId, String sessionId, String filename, int version) {
+    String key;
+    if (filename != null && filename.startsWith("user:")) {
+      key = String.format("%s/%s/user/%s/%d", appName, userId, filename, version);
+    } else {
+      key = String.format("%s/%s/%s/%s/%d", appName, userId, sessionId, filename, version);
+    }
+
+    if (s3BasePath != null && !s3BasePath.isBlank()) {
+      // Ensure the base path doesn't end with a slash to avoid double slashes
+      String cleanBasePath =
+          s3BasePath.endsWith("/") ? s3BasePath.substring(0, s3BasePath.length() - 1) : s3BasePath;
+      return cleanBasePath + "/" + key;
+    }
+    return key;
+  }
+
+  private static @Nullable String getenvOrNull(String key) {
+    String value = System.getenv(key);
+    return value == null || value.isBlank() ? null : value;
+  }
+
+  private static S3Client buildS3Client() {
+    var builder = S3Client.builder();
+    String region = getenvOrNull("S3_REGION");
+    if (region != null) {
+      builder.region(Region.of(region));
+    }
+    String endpoint = getenvOrNull("S3_ENDPOINT");
+    if (endpoint != null) {
+      builder.endpointOverride(URI.create(endpoint));
+    }
+    String pathStyle = getenvOrNull("S3_PATH_STYLE");
+    if (pathStyle != null && Boolean.parseBoolean(pathStyle)) {
+      builder.serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build());
+    }
+    return builder.build();
   }
 
   /** Closes the database connection pool. Call this when shutting down the application. */
   public void close() {
     dbHelper.close();
+    if (s3Client != null) {
+      s3Client.close();
+    }
   }
 }