feat(kits): add firestore-incremental-capture - #2938
Draft
cabljac wants to merge 10 commits into
Draft
Conversation
Wiz Scan Summary
To detect these findings earlier in the dev lifecycle, try the Wiz Code extension for VS Code, JetBrains, or Visual Studio. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request implements the Firestore Incremental Backup Stream Firebase Extension as a deployable npm package, adding a Java-based Apache Beam restoration pipeline, a setup script, and TypeScript handlers to capture and queue Firestore writes to BigQuery. The review feedback highlights several critical issues in the restoration pipeline, including a potential NullPointerException on nullable BigQuery fields, a doubled prefix path bug, and missing support for primitive arrays, NULL, and BINARY types. Additionally, the reviewer pointed out a timezone shift bug in date parsing, a potential serialization error with bigint fields, an unused and broken class, and a shell script logging improvement.
|
|
||
| GenericRecord record = schemaAndRecord.getRecord(); | ||
|
|
||
| String data = record.get("afterData").toString(); |
Contributor
There was a problem hiding this comment.
The afterData field in the BigQuery schema is nullable. If a record has a null afterData value, calling record.get("afterData").toString() will throw a NullPointerException. We should check for null and default to an empty JSON object string "{}".
Suggested change
| String data = record.get("afterData").toString(); | |
| Object afterDataObj = record.get("afterData"); | |
| String data = afterDataObj != null ? afterDataObj.toString() : "{}"; |
Comment on lines
+117
to
+118
| Document doc = Document.newBuilder().putAllFields((Map<String, Value>) firestoreMap).setName(createDocumentName( | ||
| documentPath, projectId, databaseId)).build(); |
Contributor
There was a problem hiding this comment.
Calling createDocumentName on documentPath here results in a doubled prefix (e.g., projects/.../databases/.../documents/projects/...) because documentPath was already fully constructed with the prefix on line 106. Using documentPath directly as the name fixes this restoration gap.
Document doc = Document.newBuilder().putAllFields((Map<String, Value>) firestoreMap).setName(documentPath).build();
Comment on lines
+151
to
+163
| private static List<Value> buildFirestoreList(JsonArray arr, String projectId, String databaseId) { | ||
|
|
||
| List<Value> lst = new ArrayList<>(); | ||
| for (JsonElement el : arr) { | ||
| Map<String, Value> mapData = buildFirestoreMap(el, projectId, databaseId); | ||
| Value val = Value.newBuilder().setMapValue( | ||
| MapValue.newBuilder().putAllFields(mapData)).build(); | ||
|
|
||
| lst.add(val); | ||
| } | ||
|
|
||
| return lst; | ||
| } |
Contributor
There was a problem hiding this comment.
The buildFirestoreList method currently assumes all array elements are maps, causing primitive arrays (like [1, 2]) to restore as empty maps. We can fix this restoration gap by checking if the element is a wrapped primitive and parsing it correctly.
private static List<Value> buildFirestoreList(JsonArray arr, String projectId, String databaseId) {
List<Value> lst = new ArrayList<>();
for (JsonElement el : arr) {
if (el.isJsonObject() && el.getAsJsonObject().has("type") && el.getAsJsonObject().has("value")) {
JsonObject temp = new JsonObject();
temp.add("dummy", el);
Map<String, Value> mapData = buildFirestoreMap(temp, projectId, databaseId);
if (mapData.containsKey("dummy")) {
lst.add(mapData.get("dummy"));
}
} else {
Map<String, Value> mapData = buildFirestoreMap(el, projectId, databaseId);
Value val = Value.newBuilder().setMapValue(
MapValue.newBuilder().putAllFields(mapData)).build();
lst.add(val);
}
}
return lst;
}
Comment on lines
+42
to
+47
| if (daysDiff > 7) { | ||
| // Set the date representing 7 days before the "now" date | ||
| return Instant.parse(now.minusDays(7).toString("yyyy-MM-dd'T'HH:mm:ss.SSS")); | ||
| } | ||
|
|
||
| return Instant.parse(providedDate.toString("yyyy-MM-dd'T'HH:mm:ss.SSS")); |
Contributor
There was a problem hiding this comment.
Formatting the local DateTime to a string without timezone offset and then parsing it via Instant.parse shifts the time by the local timezone offset (since Instant.parse assumes UTC). Using .toInstant() directly is more efficient, cleaner, and completely avoids timezone shift bugs.
Suggested change
| if (daysDiff > 7) { | |
| // Set the date representing 7 days before the "now" date | |
| return Instant.parse(now.minusDays(7).toString("yyyy-MM-dd'T'HH:mm:ss.SSS")); | |
| } | |
| return Instant.parse(providedDate.toString("yyyy-MM-dd'T'HH:mm:ss.SSS")); | |
| if (daysDiff > 7) { | |
| // Set the date representing 7 days before the "now" date | |
| return now.minusDays(7).toInstant(); | |
| } | |
| return providedDate.toInstant(); |
| return { type: "map", value: serializeDocument(value) }; | ||
| } | ||
|
|
||
| return { type: typeof value as SerializedType, value }; |
Contributor
There was a problem hiding this comment.
If a document contains a bigint field, typeof value will be "bigint" and the raw bigint will be returned. This will cause JSON.stringify to throw a TypeError: Do not know how to serialize a BigInt when serializing the changelog row. Converting the bigint to a string first prevents this runtime error.
if (typeof value === "bigint") {
return { type: "bigint", value: value.toString() };
}
return { type: typeof value as SerializedType, value };
Comment on lines
+138
to
+141
| default: | ||
| val = null; | ||
| continue; | ||
| } |
Contributor
There was a problem hiding this comment.
The switch statement is missing cases for NULL and BINARY types, which are documented as restoration gaps in the README. Adding support for these types ensures they are not silently dropped during restoration.
case "NULL":
val = Value.newBuilder().setNullValue(com.google.protobuf.NullValue.NULL_VALUE).build();
break;
case "BINARY":
byte[] decoded = java.util.Base64.getDecoder().decode(entryValueObject.get("value").getAsString());
val = Value.newBuilder().setBytesValue(com.google.protobuf.ByteString.copyFrom(decoded)).build();
break;
default:
val = null;
continue;
}
Comment on lines
+70
to
+71
| step() { echo -e "\n${YELLOW}==> $*${NC}"; } | ||
| ok() { echo -e "${GREEN} $*${NC}"; } |
Contributor
There was a problem hiding this comment.
Redirecting step and ok output to stderr is a best practice in shell scripting. This ensures that stdout can be cleanly used for returning values from functions (like build_jar), avoiding fragile tail -n 1 parsing hacks.
Suggested change
| step() { echo -e "\n${YELLOW}==> $*${NC}"; } | |
| ok() { echo -e "${GREEN} $*${NC}"; } | |
| step() { echo -e "\n${YELLOW}==> $*${NC}" >&2; } | |
| ok() { echo -e "${GREEN} $*${NC}" >&2; } |
Comment on lines
+27
to
+63
| public class WriteToFirestoreDoFn extends DoFn<Document, BatchWriteWithSummary> { | ||
| private final String projectId; | ||
| private final String databaseId; | ||
| private transient Firestore firestore; | ||
|
|
||
| public WriteToFirestoreDoFn(String projectId, String databaseId) { | ||
| this.projectId = projectId; | ||
| this.databaseId = databaseId; | ||
| } | ||
|
|
||
| @Setup | ||
| public void setup() { | ||
| // Initialize the Firestore client with the specified projectId | ||
| firestore = FirestoreOptions.newBuilder() | ||
| .setProjectId(projectId) | ||
| .setDatabaseId(databaseId) | ||
| .build() | ||
| .getService(); | ||
| } | ||
|
|
||
| @ProcessElement | ||
| public void processElement(ProcessContext c) { | ||
| // Write the document data to Firestore | ||
| c.output(FirestoreIO.v1().write().batchWrite().build()); | ||
| } | ||
|
|
||
| @Teardown | ||
| public void teardown() { | ||
| if (firestore != null) { | ||
| try { | ||
| firestore.close(); | ||
| } catch (Exception e) { | ||
| e.printStackTrace(); | ||
| } | ||
| } | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
This file appears to be unused in the restoration pipeline and contains compilation/runtime errors in processElement (it attempts to output a PTransform instead of a BatchWriteWithSummary). Consider deleting this file if it is not needed.
Beam/Java restoration pipeline copied verbatim from GoogleCloudPlatform/firebase-extensions@68ef3fa (firestore-incremental-capture-pipeline). The prebuilt target/restore-firestore.jar is left behind; the kit builds the jar from source.
Temporary reference copy of the firestore-incremental-capture extension from GoogleCloudPlatform/firebase-extensions@68ef3fa, formatted to repo prettier style. Removed once the kit is written, as with the other kits.
Replaces the skeleton with the five functions the extension actually deployed: the capture pair (Firestore trigger to task queue to BigQuery), the restoration pair (HTTP trigger to task queue to Dataflow), and a lifecycle task that provisions the changelog dataset and table. The three functions the extension exported but never wired into extension.yaml are dropped, along with their only callers - the Cloud Build template staging path could not work from the functions runtime, which has no gcloud. scripts/setup.sh takes over those prerequisites and builds the pipeline jar from the vendored source rather than downloading a prebuilt one. Fixes carried over from the extension: - The HTTP restoration guard compared a seconds-epoch timestamp against Date.now() in milliseconds, so it never rejected a future timestamp. Both entry points now share one validator. - DocumentReference values were tagged 'documentReference', which FirestoreReconstructor does not recognise, so references were dropped on restore. They are now tagged 'reference' to match the pipeline. - The default bucket fell back to <project>.appspot.com, wrong for projects created after September 2024. The restoration endpoint remains unauthenticated, as in the extension. The exposure is documented in the README. firebase-functions is imported through narrow subpaths: the top-level and v2 barrels pull in the RTDB provider, which fails to load because @firebase/database-compat needs @firebase/app and npm does not install it.
Findings from an audit of the migration.
The array encoding was a regression introduced by the rewrite, not an
inherited gap. FirestoreReconstructor.buildFirestoreList passes each
element straight to buildFirestoreMap, which reads field names at the top
level, so a map element must be a bare field map. The extension emitted
exactly that; the hand-rolled recursion wrapped elements in a
{type:"map"} envelope, which restores as an empty map. Arrays of maps
were the one array shape that worked.
tests/wire-format.test.ts pins the format against the extension's own
serializer tests, which are the authoritative record and are about to be
deleted with legacy/. It fails on the pre-fix encoding.
The IAM setup targeted an identity the functions do not run as.
requiresRole makes the CLI provision a managed runtime service account,
so setup.sh could not grant to it - the account does not exist until
first deploy. iam.serviceAccountUser, needed to act as the Dataflow
worker when launching a flex template, was granted only by the script and
so reached nothing; every restoration would have failed with
PERMISSION_DENIED. It moves to requiresRole, and the script now grants
the separate roles the Dataflow worker itself needs.
Also:
- Drop the DATABASE param. RestorationPipeline reads its PITR baseline
from the default database, so a non-default source was captured to the
changelog but silently absent from the restored baseline.
- Require BUCKET_NAME rather than guessing .firebasestorage.app. The
entry point reads the project's real default bucket, so it agrees with
the bucket setup.sh stages to on pre-2024 projects.
- Derive the restoration run id from the target timestamp. A retry after
a failed post-launch status write no longer starts a second Dataflow
job writing over the backup database concurrently.
- Reject an empty BACKUP_INSTANCE_ID.
- Construct the Dataflow client on first use, off the capture path.
- Enable the compute and storage APIs; workers are Compute Engine VMs.
- README: document the single-collection and default-database limits, the
required IAM, and two further restoration gaps (id-not-path collision
in the replay query, and arrays of maps surviving where primitives do
not). Add .env.example.
The kit is written and the wire format it shares with the Dataflow pipeline is pinned by tests/wire-format.test.ts, transcribed from the reference copy's own serializer tests. Recoverable from 43df05ea. Also drops the now-dead legacy entry from the firebase.json deploy ignore list.
Rebasing onto the current kits branch surfaced a deploy-model change the implementation predated. A kit stanza deploys every export as kit-<instance id>-<export name>, so the two-hop capture path was enqueueing onto task queues that do not exist: locations/<region>/functions/syncChangelogTask rather than locations/<region>/functions/kit-default-syncChangelogTask. Every captured write would have failed to enqueue. queueName() now builds the deployed name, and INSTANCE_ID becomes required with no default, since it has to match this instance's key in the instances map and a wrong value fails the same silent way. setup.sh defaults it to "default" so the flex template it stages is the one the function launches. Conventions the branch settled on since: - Drop firebase.json, .gitignore, .env.example and vitest.config.ts. Kits ship four files plus src/tests; kits/.gitignore covers build output, and its .env.* pattern is why .env.example files went. The vitest config only existed to exclude legacy/, which is gone. - firebase-functions 7.3.2, build via tsc -b, and typescript/@types/node from the root rather than per-kit devDependencies. - Restructure the README to the standard section order, and document the kit stanza, the kit-<instance id>- naming and multiple instances.
JSON.stringify throws a TypeError on a BigInt, so a document with one failed handleDocumentWrite outright rather than losing a single field, and the Firestore trigger retried it forever. The tag was already declared in SerializedType; only the encoding was missing. Introduced by this PR, not inherited from the extension.
build_jar returns the jar path on stdout, but step/ok and Maven wrote there too, so main() parsed it back out with tail -n 1. Progress and Maven output now go to stderr and the parsing hack is gone. Introduced by this PR, not inherited from the extension.
Pre-existing: the pipeline was vendored verbatim from GoogleCloudPlatform/firebase-extensions@68ef3fa, and Beam 2.51.0 (October 2023) is where all 29 high and 4 medium Wiz vulnerability findings on this PR come from. Its 195-dependency tree carried commons-compress 1.8.1 (2014), avro 1.8.2 (2017), jackson-core/mapper-asl 1.9.13 (Jackson 1.x, end of life), snakeyaml 1.33, netty 4.1.87 and protobuf-java 3.23.2. None of it is a porting regression; the npm side reports no high or critical findings. 2.75.0 moves commons-compress to 1.26.2, avro to 1.12.0, snakeyaml to 2.2, protobuf-java to 4.33.2, netty to 4.1.132 and jackson-databind to 2.18.6, and drops the Jackson 1.x artifacts from the tree entirely. The version is now a single beam.version property. Compiler source/target moves from 1.8 to 11, since Beam requires 11 and the flex template already builds on the JAVA11 base image. Verified by build only: all 7 sources compile and the shaded jar links against 2.75.0. Nothing here has been run on Dataflow, so a semantic change across the 24 minor versions would not have been caught.
The bucket-detection fix coupled the capture path to Cloud Storage. The
lazy context is shared by all five functions, and it resolved the bucket
eagerly via getStorage().bucket().name, which throws on a project that
never enabled Storage. That took down syncData, syncChangelogTask and
initIncrementalCapture, none of which need a bucket, and with retry: true
on the trigger every write then retried for days against a permanent
failure. The extension captured fine on such projects: its capture path
never read the bucket, and its lookup fell back rather than throwing.
bucketName and flexTemplatePath are now optional. Resolution no longer
throws when the bucket is unknown; the launcher raises an actionable error
instead, so the failure lands on the restoration that needs it.
Also from the same review:
- Give syncChangelogTask and initIncrementalCapture the 512MiB/540s the
extension allotted them. The v2 defaults of 256MiB/60s were a silent
downgrade, and creating a BigQuery dataset and table can outlast 60s.
- Document that a Timestamp, GeoPoint, DocumentReference or Buffer sitting
directly in an array does not survive restoration. The extension
preserved these as maps of their internals, restoring the wrong type but
keeping the data.
- Correct doc comments that still advertised {document=**} capture, which
the trigger cannot deploy. Consumers of ./lib see the comment, not the
README.
cabljac
force-pushed
the
kits-firestore-incremental-capture
branch
from
August 11, 2026 20:01
70c7ba1 to
908576e
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Ports
firestore-incremental-capturefrom GoogleCloudPlatform/firebase-extensions to a kit, and vendors its Beam/Java restoration pipeline.Five functions:
syncData→syncChangelogTask(capture),onHttpRunRestoration→runRestorationTask(restore),initIncrementalCapture(provisioning). The three the extension exported but never wired intoextension.yamlare dropped.scripts/setup.shcovers the prerequisites that need gcloud and Maven, and builds the jar from source instead of downloading it.Fixes carried over from the extension
Date.now()in ms, so it never fired.DocumentReferencewas taggeddocumentReference, whichFirestoreReconstructordoes not match, so references were dropped on restore. Nowreference..appspot.com, wrong for projects created after Sept 2024.tests/wire-format.test.tspins the TS↔Java wire format against the extension's own serializer tests.Please look at
kit-<instance id>-queue naming are reasoned from the docs and the sibling kits, not observed.