Skip to content

Commit

Permalink
Merge pull request #54 from alan-francis/main
Browse files Browse the repository at this point in the history 
Upgrade version to  0.77.0
  • Loading branch information
@ratheesh-kr
ratheesh-kr authored Jun 25, 2024
2 parents b888ad2 + bc349b6 commit 6c8b420
Show file tree
Hide file tree
Showing 6 changed files with 182 additions and 4 deletions.
Binary file modified hub-prime/lib/techbd-udi-jooq-ingress.auto.jar
View file
Binary file not shown.
2 changes: 1 addition & 1 deletion hub-prime/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
</parent>
<groupId>org.techbd</groupId>
<artifactId>hub-prime</artifactId>
<version>0.76.0</version>
<version>0.77.0</version>
<packaging>war</packaging>
<name>TechBD Hub (Prime)</name>
<description>TechBD Hub (Primary)</description>
Expand Down
88 changes: 86 additions & 2 deletions hub-prime/src/main/java/org/techbd/orchestrate/sftp/SftpManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
import java.util.Date;
import java.util.List;
import java.util.Optional;
import com.fasterxml.jackson.databind.JsonNode;
import org.techbd.conf.Configuration;

import org.apache.commons.vfs2.FileObject;
import org.apache.commons.vfs2.FileSystemException;
Expand Down Expand Up @@ -44,6 +46,27 @@ public Optional<SftpAccountsOrchctlConfig.SftpAccount> configuredTenant(final St
public record TenantSftpEgressSession(String tenantId, String sessionId, Date cachedAt,
String sessionJsonPath, String sessionJson, Date sessionFinalizedAt,
Exception error) {
public String getSessionId() {
return sessionId;
}

public Integer getFhirCount() {
try {
JsonNode jsonNode = Configuration.objectMapper.readTree(sessionJson);
JsonNode publishFhirResultNode = jsonNode.path("publishFhirResult");
if (publishFhirResultNode.isArray()) {
return publishFhirResultNode.size();
}
} catch (Exception e) {
e.printStackTrace();
}
return 0;
}
}

public record IndividualTenantSftpEgressSession(String tenantId, String sessionId, Date cachedAt,
String sessionJsonPath, String sessionJson, Date sessionFinalizedAt,
Exception error) {
}

public record TenantSftpEgressContent(String tenantId, String sftpUri, Date cachedAt, FileObject home,
Expand All @@ -53,11 +76,11 @@ public record TenantSftpEgressContent(String tenantId, String sftpUri, Date cach
static private final Logger LOG = LoggerFactory.getLogger(TenantSftpEgressContent.class);

public Optional<ZonedDateTime> mostRecentEgress() {
if(error != null) {
if (error != null) {
LOG.error("Unable to obtain most recent egress for %s".formatted(tenantId), error);
return Optional.empty();
}

try {
if (directories.length > 0) {
final var mostRecent = directories[0];
Expand Down Expand Up @@ -99,6 +122,67 @@ public TenantSftpEgressContent tenantEgressContent(final @NonNull SftpAccountsOr
return TenantSftpEgressContent.forTenant(account);
}

@Cacheable(TENANT_EGRESS_SESSIONS_CACHE_KEY)
public Optional<IndividualTenantSftpEgressSession> getTenantEgressSession(String tenantId, String InteractionId) {
final var configuredAccounts = configuredTenants.getOrchctlts();
if (configuredAccounts != null) {
for (var a : configuredAccounts) {
final var tec = tenantEgressContent(a);

if (tec.error() == null) {
if(tec.tenantId.equals(tenantId) ){
try {
for (var egressSessionDir : tec.directories()) {
if (egressSessionDir.getName().getPath().contains(InteractionId)) {
FileObject sessionJsonFile;
try {
sessionJsonFile = egressSessionDir.resolveFile("session.json");
} catch (Exception e) {
// this usually means that the SFTP directory is not a session path
// so this is not an error
continue;
}
try {
var sessionJson = sessionJsonFile.getContent();
// JSONObject sessionJsonObject = (JSONObject) sessionJson;
return Optional.of(new IndividualTenantSftpEgressSession(tec.tenantId(),
egressSessionDir.getName().getBaseName(),
tec.cachedAt(),
sessionJsonFile.getPublicURIString(),
sessionJson.getString(Charset.defaultCharset()),
Date.from(Instant.ofEpochMilli(sessionJson.getLastModifiedTime())),
null));
} catch (Exception e) {
return Optional.of(new IndividualTenantSftpEgressSession(tec.tenantId(),
egressSessionDir.getName().getBaseName(), tec.cachedAt(),
sessionJsonFile.getPublicURIString(),
null,
Date.from(Instant
.ofEpochMilli(
egressSessionDir.getContent().getLastModifiedTime())),
new RuntimeException("Unable to read session.json from %s"
.formatted(sessionJsonFile.getPublicURIString()), e)));
}
}
}
} catch (Exception e) {
return Optional
.of(new IndividualTenantSftpEgressSession(tec.tenantId(), null, tec.cachedAt(),
tec.home().getPublicURIString(),
null, null, e));
}

}
} else {
return Optional.of(new IndividualTenantSftpEgressSession(tenantId, null, tec.cachedAt(), null,
null, null,
new RuntimeException("Invalid SFTP account2 %s".formatted(tenantId), tec.error())));
}
}
}
return Optional.empty();
}

@Cacheable(TENANT_EGRESS_SESSIONS_CACHE_KEY)
public List<TenantSftpEgressSession> tenantEgressSessions() {
final var result = new ArrayList<TenantSftpEgressSession>();
Expand Down
2 changes: 1 addition & 1 deletion hub-prime/src/main/java/org/techbd/service/http/InteractionsFilter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ protected void doFilterInternal(final @NonNull HttpServletRequest origRequest,
rihr.setInteractionId(rre.interactionId().toString());
rihr.setNature(Configuration.objectMapper.valueToTree(
Map.of("nature", RequestResponseEncountered.class.getName(), "tenant_id",
tenant == null ? "N/A" : tenant.tenantId())));
tenant != null ? tenant.tenantId() != null ? tenant.tenantId(): "N/A" : "N/A")));
rihr.setContentType(MimeTypeUtils.APPLICATION_JSON_VALUE);
rihr.setInteractionKey(requestURI);
rihr.setPayload((Configuration.objectMapper
Expand Down
66 changes: 66 additions & 0 deletions hub-prime/src/main/java/org/techbd/service/http/hub/prime/ux/InteractionsController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.Optional;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -17,6 +19,7 @@
import org.springframework.web.bind.annotation.ResponseBody;
import org.techbd.conf.Configuration;
import org.techbd.orchestrate.sftp.SftpManager;
import org.techbd.orchestrate.sftp.SftpManager.TenantSftpEgressSession;
import org.techbd.service.http.aggrid.ServerRowsRequest;
import org.techbd.service.http.aggrid.ServerRowsResponse;
import org.techbd.service.http.aggrid.SqlQueryBuilder;
Expand Down Expand Up @@ -119,4 +122,67 @@ public Object httpInteraction(final @PathVariable String interactionId) {
public List<?> observeRecentSftpInteractions() {
return sftpManager.tenantEgressSessions();
}


@GetMapping("/interactions/orchctl")
@RouteMapping(label = "CSV via SFTP (egress)")
public String orchctl(final Model model, final HttpServletRequest request) {
return presentation.populateModel("page/interactions/orchctl", model, request);
}

@Operation(summary = "Recent SFTP Interactions")
@GetMapping("/support/interaction/orchctl/{tenantId}/{interactionId}.json")
@ResponseBody
public Optional<SftpManager.IndividualTenantSftpEgressSession> observeRecentSftpInteractionsWithId(final @PathVariable String tenantId, final @PathVariable String interactionId) {
return sftpManager.getTenantEgressSession(tenantId,interactionId);
}

@Operation(summary = "SFTP Interactions for Populating Grid")
@PostMapping(value = "/support/interaction/sftpExplorer.json", consumes = MediaType.APPLICATION_JSON_VALUE)
@ResponseBody
public ServerRowsResponse sftpInteractions(final @RequestBody @Nonnull ServerRowsRequest payload) {
// TODO: figure out how to write dynamic queries in jOOQ
// final var DSL = udiPrimeJpaConfig.dsl();
// final var result =
// DSL.selectFrom(INTERACTION_HTTP).offset(payload.getStartRow())
// .limit(payload.getEndRow() - payload.getStartRow() + 1).fetch();
// return ServerRowsResponse.createResponse(payload, result.intoMaps(), null);

// TODO: obtain the pivot values from the DB for the requested pivot columns
// see
// https://github.com/ag-grid/ag-grid-server-side-oracle-example/src/main/java/com/ag/grid/enterprise/oracle/demo/dao/TradeDao.java
// final var pivotValues = getPivotValues(request.getPivotCols());
final Map<String, List<String>> pivotValues = Map.of();

final var DSL = udiPrimeJpaConfig.dsl();
final var result = DSL.fetch(new SqlQueryBuilder().createSql(payload, "techbd_udi_ingress.interaction_sftp",
pivotValues));
final var rows = result.intoMaps();
var sftpResult = sftpManager.tenantEgressSessions();
Map<String, TenantSftpEgressSession> sessionMap = sftpResult.stream()
.collect(Collectors.toMap(
TenantSftpEgressSession::getSessionId,
session -> session
));


for (final var row : rows) {
String sessionId = (String) row.get("session_id");
TenantSftpEgressSession session = sessionMap.get(sessionId);
if (session != null) {
row.put("published_fhir_count", session.getFhirCount());
// Add any other fields you need from TenantSftpEgressSession
}
// this is a JSONB and might be large so don't send it even if it was requested
// since we'll get it in /support/interaction/{interactionId}.json if required;
// also since SqlQueryBuilder().createSql() is custom SQL, org.jooq.JSONB type
// will not be able to be serialized by Jackson anyway.
row.remove("session_result");
}

// create response with our results
return ServerRowsResponse.createResponse(payload, rows, pivotValues);

}

}
28 changes: 28 additions & 0 deletions udi-prime/src/main/postgres/ingestion-center/001_idempotent_interaction.psql
View file
Original file line number Diff line number Diff line change
Expand Up @@ -209,3 +209,31 @@ BEGIN
END;
END;
$$;

/*******************************************************************************************
* Comprehensive view of SFTP interactions and their associated requests. *
******************************************************************************************/
DROP VIEW IF EXISTS techbd_udi_ingress.interaction_sftp CASCADE;
CREATE OR REPLACE VIEW techbd_udi_ingress.interaction_sftp AS
SELECT
"substring"(ose.ingest_src::text,
'/SFTP/([^/]+)/'::text) as qe,
os.orch_started_at as request_time,
ose.session_id,
os."version",
count(ose.ingest_src) as ingress_count,
count(ose.ingest_src) as consumed_count,
os.session_result::JSON
FROM
techbd_orch_ctl.orch_session_entry ose
JOIN techbd_orch_ctl.orch_session os on
ose.session_id = os.orch_session_id
WHERE
"substring"(ose.ingest_src::text,
'/SFTP/([^/]+)/'::text) is not null
GROUP BY
ose.session_id,
qe,
os."version",
os.orch_started_at,
os.session_result

0 comments on commit 6c8b420

Please sign in to comment.