<commit_before>-module(pushy_sql).

-include_lib("pushy_sql.hrl").

-export([
         %% node status ops
         fetch_node_statuses/1,
         fetch_node_status/2,
         create_node_status/1,
         update_node_status/1,
         %% job ops
         fetch_job/1,
         create_job/1,
         update_job/1,
         update_job_node/1,

         sql_now/0,
         statements/1
        ]).

sql_now() -> calendar:now_to_universal_time(os:timestamp()).

%% node status ops

-spec fetch_node_statuses(binary() | string()) -> {ok, list()} | {error, term()}.
fetch_node_statuses(OrgId) ->
    case sqerl:select(list_node_statuses_for_org, [OrgId]) of
        {ok, none} ->
            {ok, []};
        {ok, Response} ->
          {ok, Response};
        {error, Reason} ->
          {error, Reason}
      end.

%% TODO Figure out correct spec
-spec fetch_node_status(binary() | string(), binary() | string() ) -> {ok, list()} | {error, term()}.
fetch_node_status(OrgId, Node) ->
    case sqerl:select(list_node_status_for_org_and_node, [OrgId, Node]) of
        {ok, none} ->
            {ok, []};
        {ok, Response} ->
          {ok, Response};
        {error, Reason} ->
          {error, Reason}
      end.

-spec create_node_status(#pushy_node_status{}) -> {ok, 1} | {error, term()}.
create_node_status(#pushy_node_status{}=NodeStatus) ->
    create_object(NodeStatus).

-spec update_node_status(#pushy_node_status{}) -> {ok, 1 | not_found} | {error, term()}.
update_node_status(#pushy_node_status{status = Status,
                                      last_updated_by = LastUpdatedBy,
                                      updated_at = UpdatedAt,
                                      node_name = NodeName,
                                      org_id = OrgId}) ->
    UpdateFields = [hb_status(Status), LastUpdatedBy, UpdatedAt, OrgId, NodeName],
    do_update(update_node_status_by_orgid_name, UpdateFields).

%% job ops

fetch_job(JobId) ->
    case sqerl:select(find_job_by_id, [JobId]) of
        {ok, Rows} when is_list(Rows) ->
            {ok, job_join_rows_to_record(Rows)};
        {ok, none} ->
            {ok, not_found};
        {error, Error} ->
            {error, Error}
    end.

create_job(#pushy_job{} = Job) ->
    create_object(Job).

-spec update_job(#pushy_job{}) -> {ok, 1 | not_found} | {error, term()}.
update_job(#pushy_job{id = JobId,
                      status = Status,
                      last_updated_by = LastUpdatedBy,
                      updated_at = UpdatedAt}) ->
    UpdateFields = [job_status(Status), LastUpdatedBy, UpdatedAt, JobId],
    do_update(update_job_by_id, UpdateFields).

-spec update_job_node(#pushy_job_node{}) -> {ok, 1 | not_found} | {error, term()}.
update_job_node(#pushy_job_node{job_id = JobId,
                                node_name = NodeName,
                                org_id = OrgId,
                                status = Status,
                                updated_at = UpdatedAt}) ->
    UpdateFields = [job_status(Status), UpdatedAt, OrgId, NodeName, JobId],
    do_update(update_job_node_by_orgid_nodename_jobid, UpdateFields).

-spec create_object(Object :: pushy_object()) -> {ok, non_neg_integer()} |
                                                           {error, term()}.
%% @doc create an object given a pushy object record
create_object(#pushy_node_status{status=Status}=NodeStatus) ->
    NodeStatus1 = NodeStatus#pushy_node_status{status=hb_status(Status)},
    create_object(insert_node_status, NodeStatus1);
%% This does not exactly follow the same pattern as it needs to
%% insert a list of job_nodes into a separate table.
create_object(#pushy_job{job_nodes = JobNodes}=Job) ->
    %% We're not dispatching to the general create_object/2 because creating a job
    %% involves adding multiple rows to multiple tables. Also, we currently embed a list of
    %% job nodes in the job record; this won't play nicely with the general
    %% 'create_object' logic, which passes all record fields (in order) as parameters for a
    %% prepared statement
    case create_object(insert_job, Job) of
      ok ->
        case insert_job_nodes(JobNodes) of
          ok ->
              %% (Remember, create_object/1 should return {ok, Number})
              {ok, 1};
          {error, Reason} ->
              %% We could have potentially inserted some job node rows before the
              %% error was thrown. If we had transactions, we could just bail out here, but
              %% instead we need to do a little cleanup. Fortunately, this just means
              %% deleting all rows with the given job id.

              % delete_job_nodes(JobId),

              %% Finally, we'll pass the root cause of
              %% the failure back up
              parse_error(Reason)
          end;
      {error, Reason} ->
        parse_error(Reason)
    end.

-spec create_object(atom(), tuple() | list()) -> {ok, non_neg_integer()} | {error, term()}.
create_object(QueryName, Record) when is_atom(QueryName) ->
    case sqerl:statement(QueryName, flatten_record(Record), count) of
        {ok, N} ->
            {ok, N};
        {error, Reason} ->
            {ok, DbType} = application:get_env(sqerl, db_type),
            parse_error(DbType, Reason)
        %% FIXME: original code for create_node had the following match, but seems like
        %% crashing would be better if we get an unexpected error.
        %% Error -> Error
    end.

%% @doc Inserts job_nodes records into the database. All records are timestamped
%% with the same stamp, namely `CreatedAt`, which is a binary string in SQL date time
%% format.
%%
%% Returns 'ok' if all records are inserted without issue. Returns an error tuple on the
%% first checksum that fails to insert into the database for whatever reason. Further
%% processing of the list is abandoned at that point.
insert_job_nodes([]) ->
    ok;
insert_job_nodes([#pushy_job_node{job_id=JobId,
                                    org_id=OrgId,
                                    node_name=NodeName,
                                    status=Status,
                                    created_at=CreatedAt,
                                    updated_at=UpdatedAt}|Rest]) ->
    case sqerl:statement(insert_job_node, [JobId, OrgId, NodeName,
            job_status(Status), CreatedAt, UpdatedAt], count) of
        {ok, 1} ->
            insert_job_nodes(Rest);
        {error, Reason} ->
            {error, Reason}
    end.

%% @doc Transforms a collection of proplists representing a job / job_nodes join query
%% result and collapses them all into a single job record. There is a row for each
%% job_node. A job_node tuple is extracted from each row; job information is extracted
%% from the final row (since it's the same in every row).
%%
%% See the 'find_job_by_id' prepared query for the row "shape".
job_join_rows_to_record(Rows) ->
    job_join_rows_to_record(Rows, []).
job_join_rows_to_record([LastRow|[]], JobNodes) ->
    C = proplist_to_job_node(LastRow),
    #pushy_job{id = safe_get(<<"id">>, LastRow),
                  org_id = safe_get(<<"org_id">>, LastRow),
                  command = safe_get(<<"command">>, LastRow),
                  status = job_status(safe_get(<<"status">>, LastRow)),
                  duration = safe_get(<<"duration">>, LastRow),
                  last_updated_by = safe_get(<<"last_updated_by">>, LastRow),
                  created_at = safe_get(<<"created_at">>, LastRow),
                  updated_at = safe_get(<<"updated_at">>, LastRow),
                  job_nodes = lists:reverse([C|JobNodes])};
job_join_rows_to_record([Row|Rest], JobNodes ) ->
    C = proplist_to_job_node(Row),
    job_join_rows_to_record(Rest, [C|JobNodes]).

%% @doc Convenience function for assembling a job_node tuple from a proplist
proplist_to_job_node(Proplist) ->
    #pushy_job_node{job_id=safe_get(<<"job_id">>, Proplist),
                    org_id=safe_get(<<"org_id">>, Proplist),
                    node_name=safe_get(<<"node_name">>, Proplist),
                    status=job_status(safe_get(<<"status">>, Proplist)),
                    created_at=safe_get(<<"created_at">>, Proplist),
                    updated_at=safe_get(<<"updated_at">>, Proplist)<commit_msg>Fixup pushy_sql to return records for node_status fetches, and cleanup timestamp a bit.<commit_after>-module(pushy_sql).

-include_lib("pushy_sql.hrl").

-export([
         %% node status ops
         fetch_node_statuses/1,
         fetch_node_status/2,
         create_node_status/1,
         update_node_status/1,
         %% job ops
         fetch_job/1,
         create_job/1,
         update_job/1,
         update_job_node/1,

         sql_now/0,
         statements/1
        ]).

sql_now() -> calendar:now_to_universal_time(os:timestamp()).

%% node status ops

-spec fetch_node_statuses(binary() | string()) -> {ok, list()} | {error, term()}.
fetch_node_statuses(OrgId) ->
    case sqerl:select(list_node_statuses_for_org, [OrgId]) of
        {ok, none} ->
            {ok, []};
        {ok, Response} ->
          {ok, [node_status_row_to_record(Row) || Row <- Response]};
        {error, Reason} ->
          {error, Reason}
      end.

%% TODO Figure out correct spec
-spec fetch_node_status(binary() | string(), binary() | string() ) -> {ok, list()} | {error, term()}.
fetch_node_status(OrgId, Node) ->
    case sqerl:select(list_node_status_for_org_and_node, [OrgId, Node]) of
        {ok, none} ->
            {ok, []};
        {ok, Response} ->
            {ok, [node_status_row_to_record(Row) || Row <-Response] };
        {error, Reason} ->
            {error, Reason}
      end.

-spec create_node_status(#pushy_node_status{}) -> {ok, 1} | {error, term()}.
create_node_status(#pushy_node_status{}=NodeStatus) ->
    create_object(NodeStatus).

-spec update_node_status(#pushy_node_status{}) -> {ok, 1 | not_found} | {error, term()}.
update_node_status(#pushy_node_status{status = Status,
                                      last_updated_by = LastUpdatedBy,
                                      updated_at = UpdatedAt,
                                      node_name = NodeName,
                                      org_id = OrgId}) ->
    UpdateFields = [hb_status(Status), LastUpdatedBy, UpdatedAt, OrgId, NodeName],
    do_update(update_node_status_by_orgid_name, UpdateFields).

%% job ops

fetch_job(JobId) ->
    case sqerl:select(find_job_by_id, [JobId]) of
        {ok, Rows} when is_list(Rows) ->
            {ok, job_join_rows_to_record(Rows)};
        {ok, none} ->
            {ok, not_found};
        {error, Error} ->
            {error, Error}
    end.

create_job(#pushy_job{} = Job) ->
    create_object(Job).

-spec update_job(#pushy_job{}) -> {ok, 1 | not_found} | {error, term()}.
update_job(#pushy_job{id = JobId,
                      status = Status,
                      last_updated_by = LastUpdatedBy,
                      updated_at = UpdatedAt}) ->
    UpdateFields = [job_status(Status), LastUpdatedBy, UpdatedAt, JobId],
    do_update(update_job_by_id, UpdateFields).

-spec update_job_node(#pushy_job_node{}) -> {ok, 1 | not_found} | {error, term()}.
update_job_node(#pushy_job_node{job_id = JobId,
                                node_name = NodeName,
                                org_id = OrgId,
                                status = Status,
                                updated_at = UpdatedAt}) ->
    UpdateFields = [job_status(Status), UpdatedAt, OrgId, NodeName, JobId],
    do_update(update_job_node_by_orgid_nodename_jobid, UpdateFields).

-spec create_object(Object :: pushy_object()) -> {ok, non_neg_integer()} |
                                                           {error, term()}.
%% @doc create an object given a pushy object record
create_object(#pushy_node_status{status=Status}=NodeStatus) ->
    NodeStatus1 = NodeStatus#pushy_node_status{status=hb_status(Status)},
    create_object(insert_node_status, NodeStatus1);
%% This does not exactly follow the same pattern as it needs to
%% insert a list of job_nodes into a separate table.
create_object(#pushy_job{job_nodes = JobNodes}=Job) ->
    %% We're not dispatching to the general create_object/2 because creating a job
    %% involves adding multiple rows to multiple tables. Also, we currently embed a list of
    %% job nodes in the job record; this won't play nicely with the general
    %% 'create_object' logic, which passes all record fields (in order) as parameters for a
    %% prepared statement
    case create_object(insert_job, Job) of
      ok ->
        case insert_job_nodes(JobNodes) of
          ok ->
              %% (Remember, create_object/1 should return {ok, Number})
              {ok, 1};
          {error, Reason} ->
              %% We could have potentially inserted some job node rows before the
              %% error was thrown. If we had transactions, we could just bail out here, but
              %% instead we need to do a little cleanup. Fortunately, this just means
              %% deleting all rows with the given job id.

              % delete_job_nodes(JobId),

              %% Finally, we'll pass the root cause of
              %% the failure back up
              parse_error(Reason)
          end;
      {error, Reason} ->
        parse_error(Reason)
    end.

-spec create_object(atom(), tuple() | list()) -> {ok, non_neg_integer()} | {error, term()}.
create_object(QueryName, Record) when is_atom(QueryName) ->
    case sqerl:statement(QueryName, flatten_record(Record), count) of
        {ok, N} ->
            {ok, N};
        {error, Reason} ->
            {ok, DbType} = application:get_env(sqerl, db_type),
            parse_error(DbType, Reason)
        %% FIXME: original code for create_node had the following match, but seems like
        %% crashing would be better if we get an unexpected error.
        %% Error -> Error
    end.

%% @doc Inserts job_nodes records into the database. All records are timestamped
%% with the same stamp, namely `CreatedAt`, which is a binary string in SQL date time
%% format.
%%
%% Returns 'ok' if all records are inserted without issue. Returns an error tuple on the
%% first checksum that fails to insert into the database for whatever reason. Further
%% processing of the list is abandoned at that point.
insert_job_nodes([]) ->
    ok;
insert_job_nodes([#pushy_job_node{job_id=JobId,
                                    org_id=OrgId,
                                    node_name=NodeName,
                                    status=Status,
                                    created_at=CreatedAt,
                                    updated_at=UpdatedAt}|Rest]) ->
    case sqerl:statement(insert_job_node, [JobId, OrgId, NodeName,
            job_status(Status), CreatedAt, UpdatedAt], count) of
        {ok, 1} ->
            insert_job_nodes(Rest);
        {error, Reason} ->
            {error, Reason}
    end.

%% sequel returns seconds as float, while many erlang functions want integer
%% we trunc instead of rounding because if it's 59.6 we don't want to round to 60!
%% TODO: Move this fn to some common code
%%
trunc_date_time_to_second({{YY,MM,DD},{H,M,S}}) ->
    {{YY,MM,DD},{H,M, erlang:trunc(S)}}.

%% @doc Transforms the proplist representing a node_status query row into a record
%%
node_status_row_to_record(Row) ->
    #pushy_node_status{org_id = safe_get(<<"org_id">>, Row),
                       node_name = safe_get(<<"node_name">>, Row),
                       status = hb_status(safe_get(<<"status">>, Row)),
                       last_updated_by = safe_get(<<"last_updated_by">>, Row),
                       created_at = trunc_date_time_to_second(safe_get(<<"created_at">>, Row)),
                       updated_at = trunc_date_time_to_second(safe_get(<<"updated_at">>, Row))}.

%% @doc Transforms a collection of proplists representing a job / job_nodes join query
%% result and collapses them all into a single job record. There is a row for each
%% job_node. A job_node tuple is extracted from each row; job information is extracted
%% from the final row (since it's the same in every row).
%%
%% See the 'find_job_by_id' prepared query for the row "shape".
job_join_rows_to_record(Rows) ->
    job_join_rows_to_record(Rows, []).
job_join_rows_to_record([LastRow|[]], JobNodes) ->
    C = proplist_to_job_node(LastRow),
    #pushy_job{id = safe_get(<<"id">>, LastRow),
                  org_id = safe_get(<<"org_id">>, LastRow),
                  command = safe_get(<<"command">>, LastRow),
                  status = job_status(safe_get(<<"status">>, LastRow)),
                  duration = safe_get(<<"duration">>, LastRow),
                  last_updated_by = safe_get(<<"last_updated_by">>, LastRow),
                  created_at = safe_get(<<"created_at">>, LastRow),
                  updated_at = safe_get(<<"updated_at">>, LastRow),
                  job_nodes = lists:reverse([C|JobNodes])};
job_join_rows_to_record([Row|Rest], JobNodes ) ->
    C = proplist_to_job_node(Row),
    job_join_rows_to_record(Rest, [C|JobNodes]).

%% @doc Convenience function for assembling a job_node tuple from a proplist
proplist_to_job_node(Proplist) ->
    #pushy_job_node{job_id=safe_get(<<"job_id">>, Proplist),
                    org_id=safe_get(<<"org_id">>, Proplist),
                    node_name=safe_get(<<"node_name">>, Proplist),
                    status=job_status(safe_get(<<"status">>, Proplist)),
                    created_at=safe_get(<<"created_at">>, Proplist),
                    updated_at=safe_get(<<"updated_at">>, Proplist)
{'fraction_non_alphanumeric': 0.11808202036289175, 'fraction_numerical': 0.0025595813662476537, 'mean_word_length': 2.3805037492789847, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 161, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7539606', 'n_tokens_mistral': 5425, 'n_tokens_neox': 5224, 'n_words': 1346}